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.

5 files changed, 58 insertions, 16 deletions

diff --git a/storage/maria/ha_maria.cc b/storage/maria/ha_maria.cc
index 26ab75cbdc0..d91c91c0376 100644
--- a/storage/maria/ha_maria.cc
+++ b/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);
 }
 
diff --git a/storage/maria/ma_close.c b/storage/maria/ma_close.c
index c355f1f1def..547c4445177 100644
--- a/storage/maria/ma_close.c
+++ b/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)
     {
diff --git a/storage/maria/ma_open.c b/storage/maria/ma_open.c
index 7cefb12faae..f8f90812e51 100644
--- a/storage/maria/ma_open.c
+++ b/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);
 
diff --git a/storage/maria/ma_state.c b/storage/maria/ma_state.c
index f130da21d07..0c673ded04e 100644
--- a/storage/maria/ma_state.c
+++ b/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)
diff --git a/storage/maria/maria_def.h b/storage/maria/maria_def.h
index b1b8ae89f31..983e0fba7f4 100644
--- a/storage/maria/maria_def.h
+++ b/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 */