Bug #17827 cluster: rename of several tables in one statement, gets multiply logged

- redo log table query for log, split into separate entries per rename, to avoid duplicates, also to only log renames that relate to ndb tables

sql/ha_ndbcluster.cc

@@ -4728,13 +4728,14 @@ int ha_ndbcluster::final_drop_index(TABLE *table_arg)
 int ha_ndbcluster::rename_table(const char *from, const char *to)
 {
   NDBDICT *dict;
+  char old_dbname[FN_HEADLEN];
   char new_tabname[FN_HEADLEN];
   const NDBTAB *orig_tab;
   int result;
 
   DBUG_ENTER("ha_ndbcluster::rename_table");
   DBUG_PRINT("info", ("Renaming %s to %s", from, to));
-  set_dbname(from);
+  set_dbname(from, old_dbname);
   set_tabname(from);
   set_tabname(to, new_tabname);
 
@@ -4742,6 +4743,7 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
     DBUG_RETURN(my_errno= HA_ERR_NO_CONNECTION);
 
   Ndb *ndb= get_ndb();
+  ndb->setDatabaseName(old_dbname);
   dict= ndb->getDictionary();
   if (!(orig_tab= dict->getTable(m_tabname)))
     ERR_RETURN(dict->getNdbError());
@@ -4833,7 +4835,8 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
                                current_thd->query, current_thd->query_length,
                                m_dbname, new_tabname,
                                0, 0,
-                               SOT_RENAME_TABLE);
+                               SOT_RENAME_TABLE,
+                               old_dbname, m_tabname);
   }
   if (share)
     free_share(&share);

sql/ha_ndbcluster_binlog.cc

@@ -885,7 +885,8 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
                              const char *db, const char *table_name,
                              uint32 ndb_table_id,
                              uint32 ndb_table_version,
-                             enum SCHEMA_OP_TYPE type)
+                             enum SCHEMA_OP_TYPE type,
+                             const char *old_db, const char *old_table_name)
 {
   DBUG_ENTER("ndbcluster_log_schema_op");
   Thd_ndb *thd_ndb= get_thd_ndb(thd);
@@ -919,10 +920,19 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
     query= tmp_buf2;
     query_length= (uint) (strxmov(tmp_buf2, "drop table `",
                                   table_name, "`", NullS) - tmp_buf2);
-    // fall through
-  case SOT_CREATE_TABLE:
-    // fall through
+    if (!share)
+      get_a_share= 1;
+    break;
   case SOT_RENAME_TABLE:
+    /* redo the rename table query as is may contain several tables */
+    query= tmp_buf2;
+    query_length= (uint) (strxmov(tmp_buf2, "rename table `",
+                                  old_db, ".", old_table_name, "` to `",
+                                  db, ".", table_name, "`", NullS) - tmp_buf2);
+    if (!share)
+      get_a_share= 1;
+    break;
+  case SOT_CREATE_TABLE:
     // fall through
   case SOT_ALTER_TABLE:
     if (!share)
@@ -983,8 +993,8 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
   }
 
   Ndb *ndb= thd_ndb->ndb;
-  char old_db[FN_REFLEN];
-  strcpy(old_db, ndb->getDatabaseName());
+  char save_db[FN_REFLEN];
+  strcpy(save_db, ndb->getDatabaseName());
 
   char tmp_buf[SCHEMA_QUERY_SIZE];
   NDBDICT *dict= ndb->getDictionary();
@@ -1091,7 +1101,7 @@ end:
           
   if (trans)
     ndb->closeTransaction(trans);
-  ndb->setDatabaseName(old_db);
+  ndb->setDatabaseName(save_db);
 
   /*
     Wait for other mysqld's to acknowledge the table operation
@@ -1169,8 +1179,8 @@ ndbcluster_update_slock(THD *thd,
   const NdbError *ndb_error= 0;
   uint32 node_id= g_ndb_cluster_connection->node_id();
   Ndb *ndb= check_ndb_in_thd(thd);
-  char old_db[128];
-  strcpy(old_db, ndb->getDatabaseName());
+  char save_db[FN_HEADLEN];
+  strcpy(save_db, ndb->getDatabaseName());
 
   char tmp_buf[SCHEMA_QUERY_SIZE];
   NDBDICT *dict= ndb->getDictionary();
@@ -1292,7 +1302,7 @@ end:
                         db, table_name);
   if (trans)
     ndb->closeTransaction(trans);
-  ndb->setDatabaseName(old_db);
+  ndb->setDatabaseName(save_db);
   DBUG_RETURN(0);
 }
 

sql/ha_ndbcluster_binlog.h

@@ -95,7 +95,9 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
                              const char *db, const char *table_name,
                              uint32 ndb_table_id,
                              uint32 ndb_table_version,
-                             enum SCHEMA_OP_TYPE type);
+                             enum SCHEMA_OP_TYPE type,
+                             const char *old_db= 0,
+                             const char *old_table_name= 0);
 int ndbcluster_handle_drop_table(Ndb *ndb, const char *event_name,
                                  NDB_SHARE *share);
 void ndb_rep_event_name(String *event_name,