Auto merge

mysql-test/r/maria2.result:
  Manual merge

KNOWN_BUGS.txt

@@ -54,16 +54,7 @@ Known bugs that are planned to be fixed later
 =============================================
 
 LOCK TABLES .. WRITE CONCURRENT is mainly done for testing MVCC.  Don't
-use this in production.  Things that is not working if you are using
-this on a table:
-
-- INSERT/REPLACE ... SELECT on an empty table may cause crashes or
-  wrong results if someone else is doing writes on the table during repair
-  or someone is doing selects during the repair index phase.
-
-INSERT ... SELECT, REPLACE ... SELECT and LOAD DATA are blocking
-inserts and SELECT for the table.  They should only have to do this if
-the destination is empty (as then we are using fast index rebuild).
+use this in production.
 
 Missing features that is planned to fix before Beta
 ===================================================

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/r/maria-big2.result

+create table t2(id int,a varchar(255),b varchar(255),key(a))engine=maria row_format=dynamic transactional=0;
+Table	Op	Msg_type	Msg_text
+test.t2	check	status	OK
+Table	Op	Msg_type	Msg_text
+test.t2	check	status	OK

mysql-test/r/maria-mvcc.result

@@ -6,70 +6,78 @@ Table	Create Table
 t1	CREATE TABLE `t1` (
   `i` int(11) DEFAULT NULL
 ) ENGINE=MARIA DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1
+insert into t1 values (0);
 lock tables t1 write concurrent;
 insert into t1 values (1);
 insert into t1 values (2);
-/* should see 1 and 2 */
+/* should see 0, 1 and 2 */
 select i from t1;
 i
+0
 1
 2
 select count(*) from t1;
 count(*)
-2
-/* should see nothing */
+3
+/* should see 0 */
 select i from t1;
 i
+0
 select count(*) from t1;
 count(*)
-0
+1
 lock tables t1 write concurrent;
 insert into t1 values (3);
 insert into t1 values (4);
-/* should see 3 and 4 */
+/* should see 0, 3 and 4 */
 select i from t1;
 i
+0
 3
 4
 select count(*) from t1;
 count(*)
-2
+3
 unlock tables;
 lock tables t1 write concurrent;
 insert into t1 values (5);
-/* should see 3, 4 and 5 */
+/* should see 0, 3, 4 and 5 */
 select i from t1;
 i
+0
 3
 4
 5
 select count(*) from t1;
 count(*)
-3
+4
 lock tables t1 write concurrent;
-/* should see 3, 4 */
+/* should see 0, 3, 4 */
 select i from t1;
 i
+0
 3
 4
 select count(*) from t1;
 count(*)
-2
+3
 insert into t1 values (6);
-/* Should see 1, 2, 6 */
+/* Should see 0, 1, 2, 6 */
 select i from t1;
 i
+0
 1
 2
 6
 select count(*) from t1;
 count(*)
-3
+4
 unlock tables;
 lock tables t1 write concurrent;
-/* Should see 1, 2, 3, 4 and 6 */
+/* Should see 0, 1, 2, 3, 4 and 6 */
 select i from t1;
 i
+0
 1
 2
 3
@@ -77,20 +85,22 @@ i
 6
 select count(*) from t1;
 count(*)
-5
-/* should see 3, 4, 5 */
+6
+/* should see 0, 3, 4, 5 */
 select i from t1;
 i
+0
 3
 4
 5
 select count(*) from t1;
 count(*)
-3
+4
 unlock tables;
-/* should see 1, 2, 3, 4, 5, 6 */
+/* should see 0, 1, 2, 3, 4, 5, 6 */
 select i from t1;
 i
+0
 1
 2
 3
@@ -99,11 +109,12 @@ i
 6
 select count(*) from t1;
 count(*)
-6
+7
 unlock tables;
-/* should see 1, 2, 3, 4, 5, 6 */
+/* should see 0, 1, 2, 3, 4, 5, 6 */
 select i from t1;
 i
+0
 1
 2
 3
@@ -112,21 +123,23 @@ i
 6
 select count(*) from t1;
 count(*)
-6
+7
 insert into t1 values (7);
-/* should see 3, 4, 7 */
+/* should see 0, 3, 4, 7 */
 select i from t1;
 i
+0
 3
 4
 7
 select count(*) from t1;
 count(*)
-3
+4
 unlock tables;
-/* should see 1, 2, 3, 4, 5, 6, 7 */
+/* should see 0, 1, 2, 3, 4, 5, 6, 7 */
 select i from t1;
 i
+0
 1
 2
 3
@@ -136,7 +149,7 @@ i
 7
 select count(*) from t1;
 count(*)
-7
+8
 drop table t1;
 CREATE TABLE t1 (fid INT NOT NULL AUTO_INCREMENT PRIMARY KEY,    g GEOMETRY NOT NULL,   SPATIAL KEY(g) ) transactional=1 row_format=page engine=maria;
 lock tables t1 write concurrent, t1 as t2 write concurrent;

mysql-test/r/maria.result

@@ -1865,3 +1865,38 @@ t1	CREATE TABLE `t1` (
   KEY `b` (`b`) KEY_BLOCK_SIZE=8192
 ) ENGINE=MARIA DEFAULT CHARSET=latin1 PAGE_CHECKSUM=0 KEY_BLOCK_SIZE=16384
 drop table t1;
+create table t2(a varchar(255),key(a))engine=maria row_format=dynamic transactional=0;
+insert into t2 values (repeat('o',124)), (repeat('h',226)), (repeat('i',236)),
+(repeat('l',234)), (repeat('b',13)), (repeat('g',236)), (repeat('y',205)),
+(repeat('c',99)), (repeat('g',145)), (repeat('o',131)), (repeat('e',63)),
+(repeat('q',155)), (repeat('k',87)), (repeat('i',54)), (repeat('p',84)),
+(repeat('m',119)), (repeat('c',2)), (repeat('a',174)), (repeat('g',160)),
+(repeat('t',147)), (repeat('n',107));
+insert into t2 values ('nupdjlafwfvuuvruxkyjxpmupihzgspkaybijztkeukgzzkrxmd');
+insert into t2 values (repeat('g',40)), (repeat('i',173)), (repeat('q',126)),
+(repeat('i',217)), (repeat('f',161)), (repeat('i',28)), (repeat('a',35)),
+(repeat('y',27)), (repeat('o',100)), (repeat('o',175)), (repeat('f',69)),
+(repeat('k',156)), (repeat('n',220)), (repeat('q',247)), (repeat('y',180)),
+(repeat('v',209)), (repeat('m',169)), (repeat('y',170)), (repeat('r',151)),
+(repeat('d',38)), (repeat('g',64)), (repeat('k',77)), (repeat('l',150)),
+(repeat('s',150)), (repeat('u',127)), (repeat('l',15)), (repeat('m',33)),
+(repeat('r',177)), (repeat('v',197)), (repeat('k',62)), (repeat('h',219)),
+(repeat('u',161)), (repeat('y',118)), (repeat('i',184)), (repeat('z',202)),
+(repeat('j',113)), (repeat('q',95)), (repeat('q',164)), (repeat('e',54)),
+(repeat('e',60)), (repeat('l',203)), (repeat('g',77)), (repeat('y',44)),
+(repeat('j',196)), (repeat('t',45)), (repeat('l',190)), (repeat('l',89)),
+(repeat('q',45)), (repeat('e',191)), (repeat('t',38)), (repeat('f',148)),
+(repeat('c',25)), (repeat('v',97)), (repeat('i',83)), (repeat('s',166)),
+(repeat('d',96)), (repeat('v',82)), (repeat('n',127)), (repeat('i',201)),
+(repeat('x',184)), (repeat('d',76)), (repeat('u',17)), (repeat('a',178));
+insert into t2 values ('hwvfiavnmufgbulapzrolonwxufheqymvjncnczlzcjokzqlsvmomcjzgzwzquyxpunxdmotdczocwliaprpubwaeccsulvittgizcutxxb');
+insert into t2 values (repeat('x',28)), (repeat('p',21)), (repeat('k',241)),
+(repeat('i',243)), (repeat('b',172)), (repeat('z',94)), (repeat('i',218)),
+(repeat('a',177)), (repeat('g',251)), (repeat('q',161)), (repeat('x',231)),
+(repeat('p',51)), (repeat('f',141)), (repeat('m',28)), (repeat('r',77)),
+(repeat('h',56)), (repeat('k',23)), (repeat('f',198)), (repeat('o',243)),
+(repeat('d',160)), (repeat('h',82));
+check table t2 extended;
+Table	Op	Msg_type	Msg_text
+test.t2	check	status	OK
+drop table t2;

mysql-test/r/maria2.result

@@ -30,3 +30,30 @@ select * from t1;
 i
 2
 drop table t1,t2;
+create table t1(id int, s char(1), unique(s)) engine=maria;
+insert into t1 values(1,"a") on duplicate key update t1.id=t1.id+1;
+insert into t1 values(1,"a") on duplicate key update t1.id=t1.id+1;
+insert into t1 select 1,"a" on duplicate key update t1.id=t1.id+1;
+select * from t1;
+id	s
+3	a
+replace into t1 select 1,"a";
+select * from t1;
+id	s
+1	a
+drop table t1;
+create table t1 (pk int primary key, apk int unique, data int) engine=maria;
+insert into t1 values (1, 1, 1), (4, 4, 4), (6, 6, 6);
+load data concurrent infile '../std_data_ln/loaddata5.dat' replace into table t1 fields terminated by '' enclosed by '' ignore 1 lines (pk, apk);
+select * from t1 order by pk;
+pk	apk	data
+1	1	1
+3	4	NULL
+5	6	NULL
+load data infile '../std_data_ln/loaddata5.dat' replace into table t1 fields terminated by '' enclosed by '' ignore 1 lines (pk, apk);
+select * from t1 order by pk;
+pk	apk	data
+1	1	1
+3	4	NULL
+5	6	NULL
+drop table t1;

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;

mysql-test/t/maria-mvcc.test

@@ -15,71 +15,75 @@ connection con1;
 
 create table t1 (i int) engine=maria;
 show create table t1;
+
+# versioning is disabled when table is empty, so insert a row
+insert into t1 values (0);
+
 lock tables t1 write concurrent;
 insert into t1 values (1);
 insert into t1 values (2);
-/* should see 1 and 2 */
+/* should see 0, 1 and 2 */
 select i from t1;
 select count(*) from t1;
 
 connect (con2,localhost,root,,);
 connection con2;
-/* should see nothing */
+/* should see 0 */
 select i from t1;
 select count(*) from t1;
 lock tables t1 write concurrent;
 insert into t1 values (3);
 insert into t1 values (4);
-/* should see 3 and 4 */
+/* should see 0, 3 and 4 */
 select i from t1;
 select count(*) from t1;
 unlock tables;
 lock tables t1 write concurrent;
 insert into t1 values (5);
-/* should see 3, 4 and 5 */
+/* should see 0, 3, 4 and 5 */
 select i from t1;
 select count(*) from t1;
 
 connect (con3,localhost,root,,);
 connection con3;
 lock tables t1 write concurrent;
-/* should see 3, 4 */
+/* should see 0, 3, 4 */
 select i from t1;
 select count(*) from t1;
 
 connection con1;
 insert into t1 values (6);
-/* Should see 1, 2, 6 */
+/* Should see 0, 1, 2, 6 */
 select i from t1;
 select count(*) from t1;
 unlock tables;
 lock tables t1 write concurrent;
-/* Should see 1, 2, 3, 4 and 6 */
+/* Should see 0, 1, 2, 3, 4 and 6 */
 select i from t1;
 select count(*) from t1;
 
 connection con2;
-/* should see 3, 4, 5 */
+/* should see 0, 3, 4, 5 */
 select i from t1;
 select count(*) from t1;
 unlock tables;
-/* should see 1, 2, 3, 4, 5, 6 */
+/* should see 0, 1, 2, 3, 4, 5, 6 */
 select i from t1;
 select count(*) from t1;
 
 connection con1;
 unlock tables;
-/* should see 1, 2, 3, 4, 5, 6 */
+/* should see 0, 1, 2, 3, 4, 5, 6 */
 select i from t1;
 select count(*) from t1;
 
 connection con3;
 insert into t1 values (7);
-/* should see 3, 4, 7 */
+/* should see 0, 3, 4, 7 */
 select i from t1;
 select count(*) from t1;
 unlock tables;
-/* should see 1, 2, 3, 4, 5, 6, 7 */
+/* should see 0, 1, 2, 3, 4, 5, 6, 7 */
 select i from t1;
 select count(*) from t1;
 

mysql-test/t/maria.test

@@ -1148,8 +1148,47 @@ create table t1 (a int not null, b int, key (a) key_block_size=1024, key(b) key_
 show create table t1;
 drop table t1;
 
+#
+# from bug37276_reduced_corruption.sql
+#
+create table t2(a varchar(255),key(a))engine=maria row_format=dynamic transactional=0;
+insert into t2 values (repeat('o',124)), (repeat('h',226)), (repeat('i',236)),
+(repeat('l',234)), (repeat('b',13)), (repeat('g',236)), (repeat('y',205)),
+(repeat('c',99)), (repeat('g',145)), (repeat('o',131)), (repeat('e',63)),
+(repeat('q',155)), (repeat('k',87)), (repeat('i',54)), (repeat('p',84)),
+(repeat('m',119)), (repeat('c',2)), (repeat('a',174)), (repeat('g',160)),
+(repeat('t',147)), (repeat('n',107));
+insert into t2 values ('nupdjlafwfvuuvruxkyjxpmupihzgspkaybijztkeukgzzkrxmd');
+insert into t2 values (repeat('g',40)), (repeat('i',173)), (repeat('q',126)),
+(repeat('i',217)), (repeat('f',161)), (repeat('i',28)), (repeat('a',35)),
+(repeat('y',27)), (repeat('o',100)), (repeat('o',175)), (repeat('f',69)),
+(repeat('k',156)), (repeat('n',220)), (repeat('q',247)), (repeat('y',180)),
+(repeat('v',209)), (repeat('m',169)), (repeat('y',170)), (repeat('r',151)),
+(repeat('d',38)), (repeat('g',64)), (repeat('k',77)), (repeat('l',150)),
+(repeat('s',150)), (repeat('u',127)), (repeat('l',15)), (repeat('m',33)),
+(repeat('r',177)), (repeat('v',197)), (repeat('k',62)), (repeat('h',219)),
+(repeat('u',161)), (repeat('y',118)), (repeat('i',184)), (repeat('z',202)),
+(repeat('j',113)), (repeat('q',95)), (repeat('q',164)), (repeat('e',54)),
+(repeat('e',60)), (repeat('l',203)), (repeat('g',77)), (repeat('y',44)),
+(repeat('j',196)), (repeat('t',45)), (repeat('l',190)), (repeat('l',89)),
+(repeat('q',45)), (repeat('e',191)), (repeat('t',38)), (repeat('f',148)),
+(repeat('c',25)), (repeat('v',97)), (repeat('i',83)), (repeat('s',166)),
+(repeat('d',96)), (repeat('v',82)), (repeat('n',127)), (repeat('i',201)),
+(repeat('x',184)), (repeat('d',76)), (repeat('u',17)), (repeat('a',178));
+insert into t2 values ('hwvfiavnmufgbulapzrolonwxufheqymvjncnczlzcjokzqlsvmomcjzgzwzquyxpunxdmotdczocwliaprpubwaeccsulvittgizcutxxb');
+insert into t2 values (repeat('x',28)), (repeat('p',21)), (repeat('k',241)),
+(repeat('i',243)), (repeat('b',172)), (repeat('z',94)), (repeat('i',218)),
+(repeat('a',177)), (repeat('g',251)), (repeat('q',161)), (repeat('x',231)),
+(repeat('p',51)), (repeat('f',141)), (repeat('m',28)), (repeat('r',77)),
+(repeat('h',56)), (repeat('k',23)), (repeat('f',198)), (repeat('o',243)),
+(repeat('d',160)), (repeat('h',82));
+
+check table t2 extended;
+drop table t2;
+
 --disable_result_log
 --disable_query_log
 eval set global storage_engine=$default_engine, maria_page_checksum=$default_checksum;
 --enable_result_log
 --enable_query_log
+

mysql-test/t/maria2.test

@@ -85,3 +85,26 @@ select count(*) from t1;
 select * from t1;
 drop table t1,t2;
 
+#
+# test INSERT ON DUPLICATE KEY UPDATE
+#
+
+create table t1(id int, s char(1), unique(s)) engine=maria;
+insert into t1 values(1,"a") on duplicate key update t1.id=t1.id+1;
+insert into t1 values(1,"a") on duplicate key update t1.id=t1.id+1;
+insert into t1 select 1,"a" on duplicate key update t1.id=t1.id+1;
+select * from t1;
+
+# test REPLACE SELECT
+replace into t1 select 1,"a";
+select * from t1;
+drop table t1;
+
+# test LOAD DATA INFILE REPLACE
+create table t1 (pk int primary key, apk int unique, data int) engine=maria;
+insert into t1 values (1, 1, 1), (4, 4, 4), (6, 6, 6);
+load data concurrent infile '../std_data_ln/loaddata5.dat' replace into table t1 fields terminated by '' enclosed by '' ignore 1 lines (pk, apk);
+select * from t1 order by pk;
+load data infile '../std_data_ln/loaddata5.dat' replace into table t1 fields terminated by '' enclosed by '' ignore 1 lines (pk, apk);
+select * from t1 order by pk;
+drop table t1;

sql/sql_parse.cc

@@ -127,7 +127,7 @@ bool end_active_trans(THD *thd)
     if (ha_commit(thd))
       error=1;
 #ifdef WITH_MARIA_STORAGE_ENGINE
-    ha_maria::implicit_commit(thd);
+    ha_maria::implicit_commit(thd, TRUE);
 #endif
   }
   thd->options&= ~(OPTION_BEGIN | OPTION_KEEP_LOG);
@@ -1132,6 +1132,10 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     {
       char *beginning_of_next_stmt= (char*) end_of_stmt;
 
+#ifdef WITH_MARIA_STORAGE_ENGINE
+      ha_maria::implicit_commit(thd, FALSE);
+#endif
+
       net_end_statement(thd);
       query_cache_end_of_result(thd);
       /*
@@ -1496,6 +1500,10 @@ bool dispatch_command(enum enum_server_command command, THD *thd,
     thd->mysys_var->abort= 0;
   }
 
+#ifdef WITH_MARIA_STORAGE_ENGINE
+  ha_maria::implicit_commit(thd, FALSE);
+#endif
+
   net_end_statement(thd);
   query_cache_end_of_result(thd);
 
@@ -2154,6 +2162,7 @@ mysql_execute_command(THD *thd)
     goto error;
 #endif
   }
+  break;
   case SQLCOM_SHOW_NEW_MASTER:
   {
     if (check_global_access(thd, REPL_SLAVE_ACL))

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;
 }
 
@@ -1805,9 +1812,12 @@ void ha_maria::start_bulk_insert(ha_rows rows)
        a lot of rows.
        We should not do this for only a few rows as this is slower and
        we don't want to update the key statistics based of only a few rows.
+       Index file rebuild requires an exclusive lock, so if versioning is on
+       don't do it (see how ha_maria::store_lock() tries to predict repair).
     */
     if (file->state->records == 0 && can_enable_indexes &&
-        (!rows || rows >= MARIA_MIN_ROWS_TO_DISABLE_INDEXES))
+        (!rows || rows >= MARIA_MIN_ROWS_TO_DISABLE_INDEXES) &&
+        (file->lock.type == TL_WRITE))
     {
       /**
          @todo for a single-row INSERT SELECT, we will go into repair, which
@@ -1951,6 +1961,7 @@ bool ha_maria::is_crashed() const
 
 int ha_maria::update_row(const uchar * old_data, uchar * new_data)
 {
+  DBUG_ASSERT(file->lock.type != TL_WRITE_CONCURRENT_INSERT);
   ha_statistic_increment(&SSV::ha_update_count);
   if (table->timestamp_field_type & TIMESTAMP_AUTO_SET_ON_UPDATE)
     table->timestamp_field->set_time();
@@ -1960,6 +1971,7 @@ int ha_maria::update_row(const uchar * old_data, uchar * new_data)
 
 int ha_maria::delete_row(const uchar * buf)
 {
+  DBUG_ASSERT(file->lock.type != TL_WRITE_CONCURRENT_INSERT);
   ha_statistic_increment(&SSV::ha_delete_count);
   return maria_delete(file, buf);
 }
@@ -2332,6 +2344,12 @@ int ha_maria::external_lock(THD *thd, int lock_type)
       {
         if (!trnman_decrement_locked_tables(trn))
         {
+          /*
+            OK should not have been sent to client yet (ACID).
+            This is a bit excessive, ACID requires this only if there are some
+            changes to commit (rollback shouldn't be tested).
+          */
+          DBUG_ASSERT(!thd->main_da.is_sent);
           /* autocommit ? rollback a transaction */
 #ifdef MARIA_CANNOT_ROLLBACK
           if (ma_commit(trn))
@@ -2393,9 +2411,14 @@ int ha_maria::start_stmt(THD *thd, thr_lock_type lock_type)
   be participant in the connection's transaction and so the implicit commits
   (ha_commit()) (like in end_active_trans()) will do the implicit commit
   without need to call this function which can then be removed.
+
+  @param  thd              THD object
+  @param  new_trn          if a new transaction should be created; a new
+                           transaction is not needed when we know that the
+                           tables will be unlocked very soon.
 */
 
-int ha_maria::implicit_commit(THD *thd)
+int ha_maria::implicit_commit(THD *thd, bool new_trn)
 {
 #ifndef MARIA_CANNOT_ROLLBACK
 #error this method should be removed
@@ -2404,11 +2427,34 @@ int ha_maria::implicit_commit(THD *thd)
   int error= 0;
   TABLE *table;
   DBUG_ENTER("ha_maria::implicit_commit");
+  if (!new_trn && thd->locked_tables)
+  {
+    /*
+      "we are under LOCK TABLES" <=> "we shouldn't commit".
+      As thd->locked_tables is true, we are either under LOCK TABLES, or in
+      prelocking; prelocking can be under LOCK TABLES, or not (and in this
+      latter case only we should commit).
+      Note that we come here only at the end of the top statement
+      (dispatch_command()), we are never committing inside a sub-statement./
+    */
+    enum prelocked_mode_type prelocked_mode= thd->prelocked_mode;
+    if ((prelocked_mode == NON_PRELOCKED) ||
+        (prelocked_mode == PRELOCKED_UNDER_LOCK_TABLES))
+    {
+      DBUG_PRINT("info", ("locked_tables, skipping"));
+      DBUG_RETURN(0);
+    }
+  }
   if ((trn= THD_TRN) != NULL)
   {
     uint locked_tables= trnman_has_locked_tables(trn);
     if (unlikely(ma_commit(trn)))
       error= 1;
+    if (!new_trn)
+    {
+      THD_TRN= NULL;
+      goto end;
+    }
     /*
       We need to create a new transaction and put it in THD_TRN. Indeed,
       tables may be under LOCK TABLES, and so they will start the next
@@ -2448,6 +2494,7 @@ int ha_maria::implicit_commit(THD *thd)
       }
     }
   }
+end:
   DBUG_RETURN(error);
 }
 
@@ -2461,6 +2508,7 @@ THR_LOCK_DATA **ha_maria::store_lock(THD *thd,
               (lock_type == TL_IGNORE || file->lock.type == TL_UNLOCK));
   if (lock_type != TL_IGNORE && file->lock.type == TL_UNLOCK)
   {
+    const enum enum_sql_command sql_command= thd->lex->sql_command;
     /*
       We have to disable concurrent inserts for INSERT ... SELECT or
       INSERT/UPDATE/DELETE with sub queries if we are using statement based
@@ -2469,15 +2517,34 @@ THR_LOCK_DATA **ha_maria::store_lock(THD *thd,
     */
     if (lock_type <= TL_READ_HIGH_PRIORITY &&
         !thd->current_stmt_binlog_row_based &&
-        (thd->lex->sql_command != SQLCOM_SELECT &&
-         thd->lex->sql_command != SQLCOM_LOCK_TABLES) &&
+        (sql_command != SQLCOM_SELECT &&
+         sql_command != SQLCOM_LOCK_TABLES) &&
+        (thd->options & OPTION_BIN_LOG) &&
         mysql_bin_log.is_open())
       lock_type= TL_READ_NO_INSERT;
-    else if (lock_type == TL_WRITE_CONCURRENT_INSERT &&
-             (thd->lex->sql_command == SQLCOM_INSERT_SELECT ||
-              thd->lex->sql_command == SQLCOM_REPLACE_SELECT ||
-              thd->lex->sql_command == SQLCOM_LOAD))
+    else if (lock_type == TL_WRITE_CONCURRENT_INSERT)
+    {
+      const enum enum_duplicates duplicates= thd->lex->duplicates;
+      /*
+        Explanation for the 3 conditions below, in order:
+
+        - Bulk insert may use repair, which will cause problems if other
+        threads try to read/insert to the table: disable versioning.
+        Note that our read of file->state->records is incorrect, as such
+        variable may have changed when we come to start_bulk_insert() (worse
+        case: we see != 0 so allow versioning, start_bulk_insert() sees 0 and
+        uses repair). This is prevented because start_bulk_insert() will not
+        try repair if we enabled versioning.
+        - INSERT SELECT ON DUPLICATE KEY UPDATE comes here with
+        TL_WRITE_CONCURRENT_INSERT but shouldn't because it can do
+        update/delete of a row and versioning doesn't support that
+        - same for LOAD DATA CONCURRENT REPLACE.
+      */
+      if ((file->state->records == 0) ||
+          (sql_command == SQLCOM_INSERT_SELECT && duplicates == DUP_UPDATE) ||
+          (sql_command == SQLCOM_LOAD && duplicates == DUP_REPLACE))
         lock_type= TL_WRITE;
+    }
     file->lock.type= lock_type;
   }
   *to++= &file->lock;
@@ -3083,15 +3150,9 @@ my_bool ha_maria::register_query_cache_table(THD *thd, char *table_name,
   actual_data_file_length= file->s->state.state.data_file_length;
   current_data_file_length= file->state->data_file_length;
 
-  if (file->s->non_transactional_concurrent_insert &&
-      current_data_file_length != actual_data_file_length)
-  {
-    /* Don't cache current statement. */
-    return FALSE;
-  }
-
-  /* It is ok to try to cache current statement. */
-  return TRUE;
+  /* Return whether is ok to try to cache current statement. */
+  DBUG_RETURN(!(file->s->non_transactional_concurrent_insert &&
+                current_data_file_length != actual_data_file_length));
 }
 #endif
 

storage/maria/ha_maria.h

@@ -161,5 +161,5 @@ class ha_maria :public handler
   {
     return file;
   }
-  static int implicit_commit(THD *thd);
+  static int implicit_commit(THD *thd, bool new_trn);
 };

storage/maria/ma_check.c

@@ -814,6 +814,7 @@ static int chk_index(HA_CHECK *param, MARIA_HA *info, MARIA_KEYDEF *keyinfo,
   MARIA_SHARE *share= info->s;
   char llbuff[22];
   uint diff_pos[2];
+  uchar tmp_key_buff[MARIA_MAX_KEY_BUFF];
   MARIA_KEY tmp_key;
   DBUG_ENTER("chk_index");
   DBUG_DUMP("buff", buff, _ma_get_page_used(share, buff));
@@ -870,7 +871,7 @@ static int chk_index(HA_CHECK *param, MARIA_HA *info, MARIA_KEYDEF *keyinfo,
   }
 
   info->last_key.keyinfo= tmp_key.keyinfo= keyinfo;
-  tmp_key.data=    info->lastkey_buff2;
+  tmp_key.data= tmp_key_buff;
   for ( ;; )
   {
     if (*_ma_killed_ptr(param))
@@ -3368,8 +3369,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 +3381,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

@@ -601,15 +601,6 @@ MARIA_HA *maria_open(const char *name, int mode, uint open_flags)
 	pos->null_bit=0;
 	pos->flag=0;					/* For purify */
 	pos++;
-
-        if ((share->keyinfo[i].flag & HA_NOSAME) && i != 0)
-        {
-          /*
-            We can't yet have versioning if there is more than one unique
-            key
-          */
-          versioning= 0;
-        }
       }
       for (i=0 ; i < uniques ; i++)
       {
@@ -665,7 +656,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 +665,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 +679,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`;