MDEV-7370: Server deadlocks on renaming a table for which persistent statistics exists

RENAME TABLE code tries to update EITS statistics. It hung, because
it used an index on (db_name,table_name) to find the table, and attempted
to update these values at the same time. The fix is do what SQL UPDATE
statement does when updating index that it's used for scanning:
- First, buffer the rowids of rows to be updated,
- then make the second pass to actually update the rows

Also fixed the call to rename_table_in_stat_tables() in sql_rename.cc
to pass the correct new database (before, it passed old db_name so cross-
database renames were not handled correctly).

Variant #2, with review feedback addressed.

2 files changed, 126 insertions, 3 deletions

diff --git a/sql/sql_rename.cc b/sql/sql_rename.cc
index 6496e1895fb..e0fd7005cd5 100644
--- a/sql/sql_rename.cc
+++ b/sql/sql_rename.cc
@@ -274,8 +274,9 @@ do_rename(THD *thd, TABLE_LIST *ren_table, char *new_db, char *new_table_name,
         LEX_STRING table_name= { ren_table->table_name,
                                  ren_table->table_name_length };
         LEX_STRING new_table= { (char *) new_alias, strlen(new_alias) };
+        LEX_STRING new_db_name= { (char*)new_db, strlen(new_db)};
         (void) rename_table_in_stat_tables(thd, &db_name, &table_name,
-                                           &db_name, &new_table);
+                                           &new_db_name, &new_table);
         if ((rc= Table_triggers_list::change_table_name(thd, ren_table->db,
                                                         old_alias,
                                                         ren_table->table_name,
diff --git a/sql/sql_statistics.cc b/sql/sql_statistics.cc
index c2150ba9309..e86c84040b4 100644
--- a/sql/sql_statistics.cc
+++ b/sql/sql_statistics.cc
@@ -592,6 +592,8 @@ public:
     stat_file->extra(HA_EXTRA_FLUSH);
     return FALSE;
   } 
+
+  friend class Stat_table_write_iter;
 };
 
 
@@ -1264,6 +1266,117 @@ public:
 
 };
 
+
+/*
+  An iterator to enumerate statistics table rows which allows to modify
+  the rows while reading them.
+
+  Used by RENAME TABLE handling to assign new dbname.tablename to statistic
+  rows.
+*/
+class Stat_table_write_iter
+{
+  Stat_table *owner;
+  IO_CACHE io_cache;
+  uchar *rowid_buf;
+  uint rowid_size;
+
+public:
+  Stat_table_write_iter(Stat_table *stat_table_arg)
+   : owner(stat_table_arg), rowid_buf(NULL),
+     rowid_size(owner->stat_file->ref_length)
+  {
+     my_b_clear(&io_cache);
+  }
+
+  /*
+    Initialize the iterator. It will return rows with n_keyparts matching the
+    curernt values.
+
+    @return  false - OK
+             true  - Error
+  */
+  bool init(uint n_keyparts)
+  {
+    if (!(rowid_buf= (uchar*)my_malloc(rowid_size, MYF(0))))
+      return true;
+
+    if (open_cached_file(&io_cache, mysql_tmpdir, TEMP_PREFIX,
+                         1024, MYF(MY_WME)))
+      return true;
+
+    handler *h= owner->stat_file;
+    uchar key[MAX_KEY_LENGTH];
+    uint prefix_len= 0;
+    for (uint i= 0; i < n_keyparts; i++)
+      prefix_len += owner->stat_key_info->key_part[i].store_length;
+
+    key_copy(key, owner->record[0], owner->stat_key_info,
+             prefix_len);
+    key_part_map prefix_map= (key_part_map) ((1 << n_keyparts) - 1);
+    h->ha_index_init(owner->stat_key_idx, false);
+    int res= h->ha_index_read_map(owner->record[0], key, prefix_map,
+                                  HA_READ_KEY_EXACT);
+    if (res)
+    {
+      reinit_io_cache(&io_cache, READ_CACHE, 0L, 0, 0);
+      /* "Key not found" is not considered an error */
+      return (res == HA_ERR_KEY_NOT_FOUND)? false: true;
+    }
+
+    do {
+      h->position(owner->record[0]);
+      my_b_write(&io_cache, h->ref, rowid_size);
+
+    } while (!h->ha_index_next_same(owner->record[0], key, prefix_len));
+
+    /* Prepare for reading */
+    reinit_io_cache(&io_cache, READ_CACHE, 0L, 0, 0);
+    h->ha_index_or_rnd_end();
+    if (h->ha_rnd_init(false))
+      return true;
+
+    return false;
+  }
+
+  /*
+     Read the next row.
+
+     @return
+        false   OK
+        true    No more rows or error.
+  */
+  bool get_next_row()
+  {
+    if (!my_b_inited(&io_cache) || my_b_read(&io_cache, rowid_buf, rowid_size))
+      return true; /* No more data */
+
+    handler *h= owner->stat_file;
+    /*
+      We should normally be able to find the row that we have rowid for. If we
+      don't, let's consider this an error.
+    */
+    int res= h->ha_rnd_pos(owner->record[0], rowid_buf);
+
+    return (res==0)? false : true;
+  }
+
+  void cleanup()
+  {
+    if (rowid_buf)
+      my_free(rowid_buf);
+    rowid_buf= NULL;
+    owner->stat_file->ha_index_or_rnd_end();
+    close_cached_file(&io_cache);
+    my_b_clear(&io_cache);
+  }
+
+  ~Stat_table_write_iter()
+  {
+    cleanup();
+  }
+};
+
 /*
   Histogram_builder is a helper class that is used to build histograms
   for columns
@@ -3285,25 +3398,34 @@ int rename_table_in_stat_tables(THD *thd, LEX_STRING *db, LEX_STRING *tab,
   stat_table= tables[INDEX_STAT].table;
   Index_stat index_stat(stat_table, db, tab);
   index_stat.set_full_table_name();
-  while (index_stat.find_next_stat_for_prefix(2))
+
+  Stat_table_write_iter index_iter(&index_stat);
+  if (index_iter.init(2))
+    rc= 1;
+  while (!index_iter.get_next_row())
   {
     err= index_stat.update_table_name_key_parts(new_db, new_tab);
     if (err & !rc)
       rc= 1;
     index_stat.set_full_table_name();
   }
+  index_iter.cleanup();
 
   /* Rename table in the statistical table column_stats */
   stat_table= tables[COLUMN_STAT].table;
   Column_stat column_stat(stat_table, db, tab);
   column_stat.set_full_table_name();
-  while (column_stat.find_next_stat_for_prefix(2))
+  Stat_table_write_iter column_iter(&column_stat);
+  if (column_iter.init(2))
+    rc= 1;
+  while (!column_iter.get_next_row())
   {
     err= column_stat.update_table_name_key_parts(new_db, new_tab);
     if (err & !rc)
       rc= 1;
     column_stat.set_full_table_name();
   }
+  column_iter.cleanup();
    
   /* Rename table in the statistical table table_stats */
   stat_table= tables[TABLE_STAT].table;