Fix for MDEV-6493: Assertion `table->file->stats.records > 0 || error'...

Fix for MDEV-6493: Assertion `table->file->stats.records > 0 || error' failure, or 'Invalid write' valgrind warnings, or crash on scenario with Aria table, view, LOCK TABLES

This bug only happens in case of paritioned tables used in LOCK TABLES and implicit_commit() was called
(as part of trying to execute a CREATE TABLE withing lock tables)

The problem was that Aria could not move the tables from one transaction to the new one, as thd->open_tables contained
a partitioned tables and not an Aria table.

Fix:
- Store a list of all open tables that are part of a share in share->open_tables
- In maria::implict_commit() use transaction->used_tables & share->open_tables to find out which tables
  was part of the current transaction instead of using thd->open_tables, which may contain partitioned tables.


mysql-test/suite/maria/maria_partition.result:
  Added test case
mysql-test/suite/maria/maria_partition.test:
  Added test case
storage/maria/ha_maria.cc:
  Use trn->used tables and share->open_tables to find out which tables was part of the current transaction instead of using thd->open_tables.
storage/maria/ma_close.c:
  Remove closed table from share->open_list
storage/maria/ma_open.c:
  Add table to share->open_list
storage/maria/ma_state.c:
  Added comment
storage/maria/maria_def.h:
  Added share->open_list, a list of all tables that is using this share.

mysql-test/suite/maria/maria_partition.result

@@ -33,3 +33,18 @@ insert into t1 values (2);
 select * from t2 left join t1 on (t2.a=t1.a) where t2.a='bbb';
 a	a
 drop table t1,t2;
+CREATE TABLE t1 (pk INT PRIMARY KEY) ENGINE=Aria PARTITION BY KEY() PARTITIONS 2;
+CREATE VIEW v1 AS SELECT * FROM t1;
+LOCK TABLE v1 WRITE;
+CREATE TABLE v1 (i INT);
+ERROR HY000: Table 'v1' was not locked with LOCK TABLES
+INSERT INTO v1 VALUES (1);
+UNLOCK TABLES;
+check table t1;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+SELECT * FROM t1;
+pk
+1
+drop table t1;
+drop view v1;

mysql-test/suite/maria/maria_partition.test

@@ -49,6 +49,28 @@ insert into t1 values (2);
 select * from t2 left join t1 on (t2.a=t1.a) where t2.a='bbb';
 drop table t1,t2;
 
+#
+# MDEV-6493
+# Assertion `table->file->stats.records > 0 || error'
+# failure, or 'Invalid write' valgrind warnings, or crash on scenario
+# with Aria table, view, LOCK TABLES #
+#
+
+CREATE TABLE t1 (pk INT PRIMARY KEY) ENGINE=Aria PARTITION BY KEY() PARTITIONS 2;
+CREATE VIEW v1 AS SELECT * FROM t1;
+
+LOCK TABLE v1 WRITE;
+--error 1100
+CREATE TABLE v1 (i INT);
+INSERT INTO v1 VALUES (1);
+UNLOCK TABLES;
+check table t1;
+
+SELECT * FROM t1;
+
+drop table t1;
+drop view v1;
+
 # Set defaults back
 --disable_result_log
 --disable_query_log

storage/maria/ha_maria.cc

@@ -2809,7 +2809,7 @@ int ha_maria::implicit_commit(THD *thd, bool new_trn)
   TRN *trn;
   int error;
   uint locked_tables;
-  TABLE *table;
+  MARIA_SHARE **used_tables= 0;
   DBUG_ENTER("ha_maria::implicit_commit");
   if (!maria_hton || !(trn= THD_TRN))
     DBUG_RETURN(0);
@@ -2825,7 +2825,33 @@ int ha_maria::implicit_commit(THD *thd, bool new_trn)
     DBUG_PRINT("info", ("locked_tables, skipping"));
     DBUG_RETURN(0);
   }
-  locked_tables= trnman_has_locked_tables(trn);
+  if ((locked_tables= trnman_has_locked_tables(trn)))
+  {
+    MARIA_SHARE **used_tables_end;
+    MARIA_USED_TABLES *tables;
+    /* Save locked tables so that we can move them to another transaction */
+
+    used_tables= (MARIA_SHARE**) my_malloc((locked_tables+1) * 
+                                           sizeof(MARIA_SHARE*),
+                                           MY_WME);
+    used_tables_end= used_tables;
+    if (!used_tables)
+    {
+      /* Continue using the old transaction; Should be safe in most cases */
+      error= HA_ERR_OUT_OF_MEM;
+      goto end;
+    }
+    
+    for (tables= (MARIA_USED_TABLES*) trn->used_tables;
+         tables;
+         tables= tables->next)
+    {
+      if (tables->share->base.born_transactional)
+        *used_tables_end++= tables->share;
+    }
+    *used_tables_end= 0;                        // End marker
+  }
+
   error= 0;
   if (unlikely(ma_commit(trn)))
     error= 1;
@@ -2858,19 +2884,28 @@ int ha_maria::implicit_commit(THD *thd, bool new_trn)
     when we should call _ma_setup_live_state() and in some cases, like
     in check table, we use the table without calling start_stmt().
   */
-  for (table=thd->open_tables; table ; table=table->next)
+  if (used_tables)
   {
-    if (table->db_stat && table->file->ht == maria_hton)
+    MARIA_SHARE **tables;
+    for (tables= used_tables; *tables ; tables++)
     {
-      MARIA_HA *handler= ((ha_maria*) table->file)->file;
-      if (handler->s->base.born_transactional)
+      MARIA_SHARE *share= *tables;
+      LIST *handlers;
+
+      /* Find table instances that was used in this transaction */
+      for (handlers= share->open_list; handlers; handlers= handlers->next)
       {
-        _ma_set_trn_for_table(handler, trn);
-        /* If handler uses versioning */
-        if (handler->s->lock_key_trees)
-        {
-          if (_ma_setup_live_state(handler))
-            error= HA_ERR_OUT_OF_MEM;
+        MARIA_HA *handler= (MARIA_HA*) handlers->data;
+        if (handler->external_ref && 
+            ((TABLE*) handler->external_ref)->in_use == thd)
+        {        
+          _ma_set_trn_for_table(handler, trn);
+          /* If handler uses versioning */
+          if (handler->s->lock_key_trees)
+          {
+            if (_ma_setup_live_state(handler))
+              error= HA_ERR_OUT_OF_MEM;
+          }
         }
       }
     }
@@ -2879,6 +2914,7 @@ int ha_maria::implicit_commit(THD *thd, bool new_trn)
   trnman_reset_locked_tables(trn, locked_tables);
 
 end:
+  my_free(used_tables);
   DBUG_RETURN(error);
 }
 

storage/maria/ma_close.c

@@ -75,7 +75,8 @@ int maria_close(register MARIA_HA *info)
     info->opt_flag&= ~(READ_CACHE_USED | WRITE_CACHE_USED);
   }
   flag= !--share->reopen;
-  maria_open_list=list_delete(maria_open_list,&info->open_list);
+  maria_open_list=  list_delete(maria_open_list,  &info->open_list);
+  share->open_list= list_delete(share->open_list, &info->share_list);
 
   my_free(info->rec_buff);
   (*share->end)(info);
@@ -86,6 +87,7 @@ int maria_close(register MARIA_HA *info)
 
     /* Check that we don't have any dangling pointers from the transaction */
     DBUG_ASSERT(share->in_trans == 0);
+    DBUG_ASSERT(share->open_list == 0);
 
     if (share->kfile.file >= 0)
     {

storage/maria/ma_open.c

@@ -207,8 +207,9 @@ static MARIA_HA *maria_clone_internal(MARIA_SHARE *share, const char *name,
   if (share->options & HA_OPTION_TMP_TABLE)
     m_info->lock.type= TL_WRITE;
 
-  m_info->open_list.data=(void*) m_info;
-  maria_open_list=list_add(maria_open_list,&m_info->open_list);
+  m_info->open_list.data= m_info->share_list.data= (void*) m_info;
+  maria_open_list=  list_add(maria_open_list,  &m_info->open_list);
+  share->open_list= list_add(share->open_list, &m_info->share_list);
 
   DBUG_RETURN(m_info);
 

storage/maria/ma_state.c

@@ -240,6 +240,7 @@ void _ma_reset_state(MARIA_HA *info)
   MARIA_STATE_HISTORY *history= share->state_history;
   DBUG_ENTER("_ma_reset_state");
 
+  /* Always true if share->now_transactional is set */
   if (history)
   {
     MARIA_STATE_HISTORY *next;
@@ -769,7 +770,7 @@ void _ma_copy_nontrans_state_information(MARIA_HA *info)
 
 /**
    Reset history
-   This is only called during repair when we the only one using the table.
+   This is only called during repair when we are the only one using the table.
 */
 
 void _ma_reset_history(MARIA_SHARE *share)

storage/maria/maria_def.h

@@ -360,6 +360,7 @@ typedef struct st_maria_share
   LEX_STRING index_file_name;
   LEX_STRING open_file_name;		/* parameter to open filename */
   uchar *file_map;			/* mem-map of file if possible */
+  LIST *open_list;			/* Tables open with this share */
   PAGECACHE *pagecache;			/* ref to the current key cache */
   MARIA_DECODE_TREE *decode_trees;
   /*
@@ -624,6 +625,7 @@ struct st_maria_handler
   PAGECACHE_FILE dfile;			/* The datafile */
   IO_CACHE rec_cache;			/* When cacheing records */
   LIST open_list;
+  LIST share_list;
   MY_BITMAP changed_fields;
   ulong row_base_length;                /* Length of row header */
   uint row_flag;                        /* Flag to store in row header */