MDEV-8200 aria bug with insert select and lock tables

This bug happens when locking the same Aria "transactional" table
(page format) more then once with LOCK TABLES and inserting into one
of them with INSERT ... SELECT when the table is empty.

Fixed by ensuring we don't use fast bulk insert if table is opened
twice with LOCK TABLES (as this changes table->s->state)

Code changes:
- Added use_count to MARIA_USED_TABLES to be able to check if
  table is opened twice for a statement/lock table
- Don't clear history or reset info->start_state if we
  don't have versioning. One reason for the bug was
  was that info->start_state was set to point to different
  states for the two tables.  If there is no versioning
  info->start_state should always point to info->s->state.common.

Other things:
- Fixed also some typos that was noticed while scanning the code
- More DBUG_PRINT

mysql-test/suite/maria/lock.result

@@ -30,3 +30,72 @@ drop table t1;
 CREATE TABLE t1 (i INT) ENGINE=Aria;
 LOCK TABLES t1 WRITE, t1 AS t1a WRITE;
 DROP TABLE t1;
+#
+#  MDEV-8200 aria bug with insert select when select is a aria table
+#  (wrong result or assertion failure:
+# `table->file->stats.records > 0 || error')
+#
+CREATE TABLE t1 (f1 INT) ENGINE=Aria;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `f1` int(11) DEFAULT NULL
+) ENGINE=Aria DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1
+INSERT INTO t1 VALUES (1);
+CREATE TABLE t2 (f2 INT) ENGINE=MyISAM;
+CREATE TABLE tmp (f3 INT) engine=Aria;
+LOCK TABLE t2 WRITE, tmp WRITE, tmp AS tmp_alias WRITE, t1 WRITE;
+INSERT INTO tmp SELECT f1 FROM t1;
+INSERT INTO t2 SELECT f3 FROM tmp AS tmp_alias;
+select * from t2;
+f2
+1
+unlock tables;
+DROP TABLE t1,t2,tmp;
+#
+# Same without transactional
+#
+CREATE TABLE t1 (f1 INT) transactional=0 ENGINE=Aria;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `f1` int(11) DEFAULT NULL
+) ENGINE=Aria DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1 TRANSACTIONAL=0
+INSERT INTO t1 VALUES (2);
+CREATE TABLE t2 (f2 INT) ENGINE=MyISAM;
+CREATE TABLE tmp (f3 INT) transactional=0 engine=Aria;
+LOCK TABLE t2 WRITE, tmp WRITE, tmp AS tmp_alias WRITE, t1 WRITE;
+INSERT INTO tmp SELECT f1 FROM t1;
+INSERT INTO t2 SELECT f3 FROM tmp AS tmp_alias;
+select * from t2;
+f2
+2
+unlock tables;
+DROP TABLE t1,t2,tmp;
+#
+# Using spatical keys (disables versioning)
+#
+CREATE TABLE t1 (f1 INT, c1 geometry NOT NULL, SPATIAL KEY i1 (c1)) transactional=1 ENGINE=Aria;
+SHOW CREATE TABLE t1;
+Table	Create Table
+t1	CREATE TABLE `t1` (
+  `f1` int(11) DEFAULT NULL,
+  `c1` geometry NOT NULL,
+  SPATIAL KEY `i1` (`c1`)
+) ENGINE=Aria DEFAULT CHARSET=latin1 PAGE_CHECKSUM=1 TRANSACTIONAL=1
+INSERT INTO t1 VALUES (3,
+PolygonFromText('POLYGON((-18.6086111000 -66.9327777000,
+                          -18.6055555000 -66.8158332999,
+                          -18.7186111000 -66.8102777000,
+                          -18.7211111000 -66.9269443999,
+                          -18.6086111000 -66.9327777000))'));
+CREATE TABLE t2 (f2 INT) ENGINE=MyISAM;
+CREATE TABLE tmp (f3 INT, c1 geometry NOT NULL, SPATIAL KEY i1 (c1)) transactional=1 ENGINE=Aria;
+LOCK TABLE t2 WRITE, tmp WRITE, tmp AS tmp_alias WRITE, t1 WRITE;
+INSERT INTO tmp SELECT f1,c1 FROM t1;
+INSERT INTO t2 (f2) SELECT f3 FROM tmp AS tmp_alias;
+select * from t2;
+f2
+3
+unlock tables;
+DROP TABLE t1,t2,tmp;

mysql-test/suite/maria/lock.test

@@ -50,3 +50,58 @@ drop table t1;
 CREATE TABLE t1 (i INT) ENGINE=Aria;
 LOCK TABLES t1 WRITE, t1 AS t1a WRITE;
 DROP TABLE t1;
+
+--echo #
+--echo #  MDEV-8200 aria bug with insert select when select is a aria table
+--echo #  (wrong result or assertion failure:
+--echo # `table->file->stats.records > 0 || error')
+--echo #
+
+CREATE TABLE t1 (f1 INT) ENGINE=Aria;
+SHOW CREATE TABLE t1;
+INSERT INTO t1 VALUES (1);
+CREATE TABLE t2 (f2 INT) ENGINE=MyISAM;
+CREATE TABLE tmp (f3 INT) engine=Aria;
+LOCK TABLE t2 WRITE, tmp WRITE, tmp AS tmp_alias WRITE, t1 WRITE;
+INSERT INTO tmp SELECT f1 FROM t1;
+INSERT INTO t2 SELECT f3 FROM tmp AS tmp_alias;
+select * from t2;
+unlock tables;
+DROP TABLE t1,t2,tmp;
+
+--echo #
+--echo # Same without transactional
+--echo #
+
+CREATE TABLE t1 (f1 INT) transactional=0 ENGINE=Aria;
+SHOW CREATE TABLE t1;
+INSERT INTO t1 VALUES (2);
+CREATE TABLE t2 (f2 INT) ENGINE=MyISAM;
+CREATE TABLE tmp (f3 INT) transactional=0 engine=Aria;
+LOCK TABLE t2 WRITE, tmp WRITE, tmp AS tmp_alias WRITE, t1 WRITE;
+INSERT INTO tmp SELECT f1 FROM t1;
+INSERT INTO t2 SELECT f3 FROM tmp AS tmp_alias;
+select * from t2;
+unlock tables;
+DROP TABLE t1,t2,tmp;
+
+--echo #
+--echo # Using spatical keys (disables versioning)
+--echo #
+
+CREATE TABLE t1 (f1 INT, c1 geometry NOT NULL, SPATIAL KEY i1 (c1)) transactional=1 ENGINE=Aria;
+SHOW CREATE TABLE t1;
+INSERT INTO t1 VALUES (3,
+PolygonFromText('POLYGON((-18.6086111000 -66.9327777000,
+                          -18.6055555000 -66.8158332999,
+                          -18.7186111000 -66.8102777000,
+                          -18.7211111000 -66.9269443999,
+                          -18.6086111000 -66.9327777000))'));
+CREATE TABLE t2 (f2 INT) ENGINE=MyISAM;
+CREATE TABLE tmp (f3 INT, c1 geometry NOT NULL, SPATIAL KEY i1 (c1)) transactional=1 ENGINE=Aria;
+LOCK TABLE t2 WRITE, tmp WRITE, tmp AS tmp_alias WRITE, t1 WRITE;
+INSERT INTO tmp SELECT f1,c1 FROM t1;
+INSERT INTO t2 (f2) SELECT f3 FROM tmp AS tmp_alias;
+select * from t2;
+unlock tables;
+DROP TABLE t1,t2,tmp;

storage/maria/ha_maria.cc

@@ -2112,11 +2112,16 @@ void ha_maria::start_bulk_insert(ha_rows rows)
        safety net for now, we don't remove the test of
        file->state->records, because there is uncertainty on what will
        happen during repair if the two states disagree.
+
+       We also have to check in case of transactional tables that the
+       user has not used LOCK TABLE on the table twice.
     */
     if ((file->state->records == 0) &&
         (share->state.state.records == 0) && can_enable_indexes &&
         (!rows || rows >= MARIA_MIN_ROWS_TO_DISABLE_INDEXES) &&
-        (file->lock.type == TL_WRITE || file->lock.type == TL_UNLOCK))
+        (file->lock.type == TL_WRITE || file->lock.type == TL_UNLOCK) &&
+        (!share->have_versioning || !share->now_transactional ||
+         file->used_tables->use_count == 1))
     {
       /**
          @todo for a single-row INSERT SELECT, we will go into repair, which

storage/maria/ma_blockrec.c

@@ -6129,7 +6129,7 @@ my_bool write_hook_for_undo_row_insert(enum translog_record_type type
 
 
 /**
-   @brief Upates "records" and calls the generic UNDO hook
+   @brief Updates "records" and calls the generic UNDO hook
 
    @return Operation status, always 0 (success)
 */

storage/maria/ma_state.c

@@ -58,6 +58,7 @@ my_bool _ma_setup_live_state(MARIA_HA *info)
   MARIA_USED_TABLES *tables;
   MARIA_STATE_HISTORY *history;
   DBUG_ENTER("_ma_setup_live_state");
+  DBUG_PRINT("enter", ("info: %p", info));
 
   DBUG_ASSERT(share->lock_key_trees);
 
@@ -110,6 +111,8 @@ my_bool _ma_setup_live_state(MARIA_HA *info)
 end:
   info->state_start= &tables->state_start;
   info->state= &tables->state_current;
+  info->used_tables= tables;
+  tables->use_count++;
 
   /*
     Mark in transaction state if we are not using transid (versioning)
@@ -118,6 +121,7 @@ my_bool _ma_setup_live_state(MARIA_HA *info)
   */
   tables->state_current.no_transid|= !(info->row_flag & ROW_FLAG_TRANSID);
 
+  DBUG_PRINT("exit", ("tables: %p  info->state: %p", tables, info->state));
   DBUG_RETURN(0);
 }
 
@@ -241,9 +245,10 @@ void _ma_reset_state(MARIA_HA *info)
   DBUG_ENTER("_ma_reset_state");
 
   /* Always true if share->now_transactional is set */
-  if (history)
+  if (history && share->have_versioning)
   {
     MARIA_STATE_HISTORY *next;
+    DBUG_PRINT("info", ("resetting history"));
 
     /* Set the current history to current state */
     share->state_history->state= share->state.state;
@@ -255,7 +260,7 @@ void _ma_reset_state(MARIA_HA *info)
       my_free(history);
     }
     share->state_history->next= 0;
-    share->state_history->trid= 0;              /* Visibile for all */
+    share->state_history->trid= 0;              /* Visible for all */
   }
   DBUG_VOID_RETURN;
 }
@@ -597,7 +602,7 @@ void _ma_remove_table_from_trnman(MARIA_SHARE *share, TRN *trn)
 
   SYNOPSIS
     _ma_get_status()
-    param		Pointer to Myisam handler
+    param		Pointer to Aria handler
     concurrent_insert	Set to 1 if we are going to do concurrent inserts
 			(THR_WRITE_CONCURRENT_INSERT was used)
 */
@@ -627,6 +632,8 @@ void _ma_block_get_status(void* param, my_bool concurrent_insert)
 my_bool _ma_block_start_trans(void* param)
 {
   MARIA_HA *info=(MARIA_HA*) param;
+  DBUG_ENTER("_ma_block_start_trans");
+
   if (info->s->lock_key_trees)
   {
     /*
@@ -634,7 +641,7 @@ my_bool _ma_block_start_trans(void* param)
       out of memory conditions)
       TODO: Fix this by having one extra state pre-allocated
     */
-    return _ma_setup_live_state(info);
+    DBUG_RETURN(_ma_setup_live_state(info));
   }
   else
   {
@@ -663,9 +670,9 @@ my_bool _ma_block_start_trans(void* param)
       Assume for now that this doesn't fail (It can only fail in
       out of memory conditions)
     */
-    return maria_create_trn_hook(info) != 0;
+    DBUG_RETURN(maria_create_trn_hook(info) != 0);
   }
-  return 0;
+  DBUG_RETURN(0);
 }
 
 
@@ -697,7 +704,7 @@ my_bool _ma_block_check_status(void *param __attribute__((unused)))
 my_bool _ma_block_start_trans_no_versioning(void* param)
 {
   MARIA_HA *info=(MARIA_HA*) param;
-  DBUG_ENTER("_ma_block_get_status_no_version");
+  DBUG_ENTER("_ma_block_start_trans_no_versioning");
   DBUG_ASSERT(info->s->base.born_transactional && !info->s->lock_key_trees);
 
   info->state->changed= 0;              /* from _ma_reset_update_flag() */
@@ -722,6 +729,8 @@ my_bool _ma_block_start_trans_no_versioning(void* param)
 void maria_versioning(MARIA_HA *info, my_bool versioning)
 {
   MARIA_SHARE *share= info->s;
+  DBUG_ENTER("maria_versioning");
+
   /* For now, this is a hack */
   if (share->have_versioning)
   {
@@ -738,6 +747,7 @@ void maria_versioning(MARIA_HA *info, my_bool versioning)
       info->state= &share->state.state;	/* Change global values by default */
     info->state_start= info->state;             /* Initial values */
   }
+  DBUG_VOID_RETURN;
 }
 
 

storage/maria/ma_state.h

@@ -34,6 +34,7 @@ typedef struct st_used_tables {
   struct st_maria_share *share;
   MARIA_STATUS_INFO state_current;
   MARIA_STATUS_INFO state_start;
+  uint use_count;
 } MARIA_USED_TABLES;
 
 

storage/maria/maria_def.h

@@ -573,6 +573,7 @@ struct st_maria_handler
   struct st_ma_transaction *trn;        /* Pointer to active transaction */
   MARIA_STATUS_INFO *state, state_save;
   MARIA_STATUS_INFO *state_start;       /* State at start of transaction */
+  MARIA_USED_TABLES *used_tables;
   MARIA_ROW cur_row;                    /* The active row that we just read */
   MARIA_ROW new_row;			/* Storage for a row during update */
   MARIA_KEY last_key;                   /* Last found key */