Bug #17838 binlog not setup on seconday master after rename

-added setting up of binlogging after receiving rename table
+cleanup


mysql-test/r/ndb_binlog_ddl_multi.result:
  Bug #17838 binlog not setup on seconday master after rename
mysql-test/t/ndb_binlog_ddl_multi.test:
  Bug #17838 binlog not setup on seconday master after rename
mysql-test/include/show_binlog_events.inc:
  New BitKeeper file ``mysql-test/include/show_binlog_events.inc''

mysql-test/include/show_binlog_events.inc

+--let $binlog_start=102
+--replace_result $binlog_start <binlog_start>
+--replace_column 2 # 4 # 5 #
+--eval show binlog events from $binlog_start

mysql-test/r/ndb_binlog_ddl_multi.result

 drop database if exists mysqltest;
-drop table if exists t1,t2;
+drop table if exists t1,t2,t3;
 drop database if exists mysqltest;
-drop table if exists t1,t2;
+drop table if exists t1,t2,t3;
 reset master;
 reset master;
 create database mysqltest;
@@ -123,3 +123,68 @@ master-bin1.000001	#	Table_map	#	#	cluster.apply_status
 master-bin1.000001	#	Write_rows	#	#	
 master-bin1.000001	#	Query	#	#	COMMIT
 master-bin1.000001	#	Query	#	#	use `test`; drop table `t1`
+reset master;
+show tables;
+Tables_in_test
+reset master;
+show tables;
+Tables_in_test
+create table t1 (a int key) engine=ndb;
+create table t2 (a int key) engine=ndb;
+create table t3 (a int key) engine=ndb;
+rename table t3 to t4, t2 to t3, t1 to t2, t4 to t1;
+show binlog events from <binlog_start>;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin1.000001	#	Query	#	#	use `test`; create table t1 (a int key) engine=ndb
+master-bin1.000001	#	Query	#	#	use `test`; create table t2 (a int key) engine=ndb
+master-bin1.000001	#	Query	#	#	use `test`; create table t3 (a int key) engine=ndb
+master-bin1.000001	#	Query	#	#	BEGIN
+master-bin1.000001	#	Table_map	#	#	cluster.apply_status
+master-bin1.000001	#	Write_rows	#	#	
+master-bin1.000001	#	Query	#	#	COMMIT
+master-bin1.000001	#	Query	#	#	use `test`; rename table `test.t3` to `test.t4`
+master-bin1.000001	#	Query	#	#	BEGIN
+master-bin1.000001	#	Table_map	#	#	cluster.apply_status
+master-bin1.000001	#	Write_rows	#	#	
+master-bin1.000001	#	Query	#	#	COMMIT
+master-bin1.000001	#	Query	#	#	use `test`; rename table `test.t2` to `test.t3`
+master-bin1.000001	#	Query	#	#	BEGIN
+master-bin1.000001	#	Table_map	#	#	cluster.apply_status
+master-bin1.000001	#	Write_rows	#	#	
+master-bin1.000001	#	Query	#	#	COMMIT
+master-bin1.000001	#	Query	#	#	use `test`; rename table `test.t1` to `test.t2`
+master-bin1.000001	#	Query	#	#	BEGIN
+master-bin1.000001	#	Table_map	#	#	cluster.apply_status
+master-bin1.000001	#	Write_rows	#	#	
+master-bin1.000001	#	Query	#	#	COMMIT
+master-bin1.000001	#	Query	#	#	use `test`; rename table `test.t4` to `test.t1`
+drop table t1;
+drop table t2;
+drop table t3;
+reset master;
+show tables;
+Tables_in_test
+reset master;
+show tables;
+Tables_in_test
+create table t1 (a int key) engine=ndb;
+insert into t1 values(1);
+rename table t1 to t2;
+insert into t2 values(2);
+show binlog events from <binlog_start>;
+Log_name	Pos	Event_type	Server_id	End_log_pos	Info
+master-bin1.000001	#	Query	#	#	use `test`; create table t1 (a int key) engine=ndb
+master-bin1.000001	#	Query	#	#	BEGIN
+master-bin1.000001	#	Table_map	#	#	cluster.apply_status
+master-bin1.000001	#	Write_rows	#	#	
+master-bin1.000001	#	Table_map	#	#	test.t1
+master-bin1.000001	#	Write_rows	#	#	
+master-bin1.000001	#	Query	#	#	COMMIT
+master-bin1.000001	#	Query	#	#	use `test`; rename table `test.t1` to `test.t2`
+master-bin1.000001	#	Query	#	#	BEGIN
+master-bin1.000001	#	Table_map	#	#	cluster.apply_status
+master-bin1.000001	#	Write_rows	#	#	
+master-bin1.000001	#	Table_map	#	#	test.t2
+master-bin1.000001	#	Write_rows	#	#	
+master-bin1.000001	#	Query	#	#	COMMIT
+drop table t2;

mysql-test/t/ndb_binlog_ddl_multi.test

@@ -5,18 +5,16 @@
 --disable_warnings
 connection server2;
 drop database if exists mysqltest;
-drop table if exists t1,t2;
+drop table if exists t1,t2,t3;
 connection server1;
 drop database if exists mysqltest;
-drop table if exists t1,t2;
+drop table if exists t1,t2,t3;
 --connection server1
 reset master;
 --connection server2
 reset master;
 --enable_warnings
 
---let $binlog_start=102
-
 #
 # basic test to see if ddl distribution works across
 # multiple binlogs
@@ -33,15 +31,10 @@ create table t1 (a int primary key) engine=ndb;
 
 --connection server2
 create table t2 (a int primary key) engine=ndb;
---replace_result $binlog_start <binlog_start>
---replace_column 2 # 4 # 5 # 
---eval show binlog events from $binlog_start
+--source include/show_binlog_events.inc
 
 --connection server1
---replace_result $binlog_start <binlog_start>
---replace_column 2 # 4 # 5 # 
---eval show binlog events from $binlog_start
-
+-- source include/show_binlog_events.inc
 
 # alter table
 --connection server1
@@ -53,9 +46,7 @@ reset master;
 alter table t2 add column (b int);
 
 --connections server1
---replace_result $binlog_start <binlog_start>
---replace_column 2 # 4 # 5 # 
---eval show binlog events from $binlog_start
+--source include/show_binlog_events.inc
 
 
 # alter database
@@ -91,9 +82,7 @@ drop database mysqltest;
 create table t1 (a int primary key) engine=ndb;
 
 --connection server2
---replace_result $binlog_start <binlog_start>
---replace_column 2 # 4 # 5 # 
---eval show binlog events from $binlog_start
+--source include/show_binlog_events.inc
 
 --connection server2
 drop table t2;
@@ -144,6 +133,51 @@ ENGINE =NDB;
 drop table t1;
 
 --connection server2
---replace_result $binlog_start <binlog_start>
---replace_column 2 # 4 # 5 # 
---eval show binlog events from $binlog_start
+--source include/show_binlog_events.inc
+
+#
+# Bug #17827 cluster: rename of several tables in one statement,
+# gets multiply logged
+#
+--connection server1
+reset master;
+show tables;
+--connection server2
+reset master;
+show tables;
+
+--connection server1
+create table t1 (a int key) engine=ndb;
+create table t2 (a int key) engine=ndb;
+create table t3 (a int key) engine=ndb;
+rename table t3 to t4, t2 to t3, t1 to t2, t4 to t1;
+--connection server2
+--source include/show_binlog_events.inc
+
+drop table t1;
+drop table t2;
+drop table t3;
+
+#
+# Bug #17838 binlog not setup on seconday master after rename
+#
+#
+--connection server1
+reset master;
+show tables;
+--connection server2
+reset master;
+show tables;
+
+--connection server1
+create table t1 (a int key) engine=ndb;
+insert into t1 values(1);
+rename table t1 to t2;
+insert into t2 values(2);
+
+# now we should see data in table t1 _and_ t2
+# prior to bug fix, data was missing for t2
+--connection server2
+--source include/show_binlog_events.inc
+
+drop table t2;

sql/ha_ndbcluster_binlog.cc

@@ -1314,22 +1314,35 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
                          NDB_SHARE *share)
 {
   DBUG_ENTER("ndb_handle_schema_change");
-  int remote_drop_table= 0, do_close_cached_tables= 0;
-  const char *dbname= share->table->s->db.str;
-  const char *tabname= share->table->s->table_name.str;
-  bool online_alter_table= (pOp->getEventType() == NDBEVENT::TE_ALTER &&
-                            pOp->tableFrmChanged());
+  bool do_close_cached_tables= FALSE;
+  bool is_online_alter_table= FALSE;
+  bool is_rename_table= FALSE;
+  bool is_remote_change=
+    (uint) pOp->getReqNodeId() != g_ndb_cluster_connection->node_id();
 
-  if (pOp->getEventType() != NDBEVENT::TE_CLUSTER_FAILURE &&
-      (uint) pOp->getReqNodeId() != g_ndb_cluster_connection->node_id())
+  if (pOp->getEventType() == NDBEVENT::TE_ALTER)
+  {
+    if (pOp->tableFrmChanged())
+    {
+      is_online_alter_table= TRUE;
+    }
+    else
+    {
+      DBUG_ASSERT(pOp->tableNameChanged());
+      is_rename_table= TRUE;
+    }
+  }
+
+  if (is_remote_change) /* includes CLUSTER_FAILURE */
   {
-    TABLE_SHARE *table_share= share->table->s;
     TABLE* table= share->table;
+    TABLE_SHARE *table_share= table->s;
+    const char *dbname= table_share->db.str;
     
     /* 
        Invalidate table and all it's indexes
     */
-    ndb->setDatabaseName(share->table->s->db.str);
+    ndb->setDatabaseName(dbname);
     Thd_ndb *thd_ndb= get_thd_ndb(thd);
     DBUG_ASSERT(thd_ndb != NULL);
     Ndb* old_ndb= thd_ndb->ndb;
@@ -1341,8 +1354,9 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
     table_handler.invalidate_dictionary_cache(TRUE);
     thd_ndb->ndb= old_ndb;
     
-    if (online_alter_table)
+    if (is_online_alter_table)
     {
+      const char *tabname= table_share->table_name.str;
       char key[FN_REFLEN];
       const void *data= 0, *pack_data= 0;
       uint length, pack_length;
@@ -1383,11 +1397,10 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
         pthread_mutex_unlock(&LOCK_open);
       }
     }
-    remote_drop_table= 1;
   }
 
   // If only frm was changed continue replicating
-  if (online_alter_table)
+  if (is_online_alter_table)
   {
     /* Signal ha_ndbcluster::alter_table that drop is done */
     (void) pthread_cond_signal(&injector_cond);
@@ -1395,6 +1408,22 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
   }
 
   (void) pthread_mutex_lock(&share->mutex);
+  if (is_rename_table && !is_remote_change)
+  {
+    DBUG_PRINT("info", ("Detected name change of table %s.%s",
+                        share->db, share->table_name));
+    /* ToDo: remove printout */
+    if (ndb_extra_logging)
+      sql_print_information("NDB Binlog: rename table %s%s/%s -> %s.",
+                            share_prefix, share->table->s->db.str,
+                            share->table->s->table_name.str,
+                            share->key);
+    /* do the rename of the table in the share */
+    share->table->s->db.str= share->db;
+    share->table->s->db.length= strlen(share->db);
+    share->table->s->table_name.str= share->table_name;
+    share->table->s->table_name.length= strlen(share->table_name);
+  }
   DBUG_ASSERT(share->op == pOp || share->op_old == pOp);
   if (share->op_old == pOp)
     share->op_old= 0;
@@ -1408,11 +1437,11 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
 
   pthread_mutex_lock(&ndbcluster_mutex);
   free_share(&share, TRUE);
-  if (remote_drop_table && share && share->state != NSS_DROPPED)
+  if (is_remote_change && share && share->state != NSS_DROPPED)
   {
-    DBUG_PRINT("info", ("remote drop table"));
+    DBUG_PRINT("info", ("remote change"));
     if (share->use_count != 1)
-      do_close_cached_tables= 1;
+      do_close_cached_tables= TRUE;
     share->state= NSS_DROPPED;
     free_share(&share, TRUE);
   }
@@ -1464,24 +1493,36 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
         int log_query= 0;
         DBUG_PRINT("info", ("log query_length: %d  query: '%s'",
                             schema->query_length, schema->query));
+        char key[FN_REFLEN];
+        build_table_filename(key, sizeof(key), schema->db, schema->name, "");
+        NDB_SHARE *share= get_share(key, 0, false, false);
+
         switch ((enum SCHEMA_OP_TYPE)schema->type)
         {
         case SOT_DROP_TABLE:
           /* binlog dropping table after any table operations */
-          if (ndb_binlog_running)
+          if (share && share->op)
             post_epoch_log_list->push_back(schema, mem_root);
           log_query= 0;
           break;
         case SOT_RENAME_TABLE:
-          /* fall through */
+          if (share && share->op)
+          {
+            log_query= 0;
+            post_epoch_log_list->push_back(schema, mem_root);
+            break; /* discovery will be handled by binlog */
+          }
+          goto sot_create_table;
         case SOT_ALTER_TABLE:
-          if (ndb_binlog_running)
+          if (share && share->op)
           {
-            log_query= 1;
+            log_query= 0;
+            post_epoch_log_list->push_back(schema, mem_root);
             break; /* discovery will be handled by binlog */
           }
-          /* fall through */
+          goto sot_create_table;
         case SOT_CREATE_TABLE:
+      sot_create_table:
           pthread_mutex_lock(&LOCK_open);
           if (ndb_create_table_from_engine(thd, schema->db, schema->name))
           {
@@ -1514,10 +1555,6 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
           break;
         case SOT_CLEAR_SLOCK:
         {
-          char key[FN_REFLEN];
-          build_table_filename(key, sizeof(key),
-                               schema->db, schema->name, "");
-          NDB_SHARE *share= get_share(key, 0, false, false);
           if (share)
           {
             pthread_mutex_lock(&share->mutex);
@@ -1565,29 +1602,20 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
                             schema->name[0] == 0 || thd->db[0] == 0);
           thd->db= thd_db_save;
         }
+        if (share)
+          free_share(&share);
       }
     }
     break;
     case NDBEVENT::TE_DELETE:
       // skip
       break;
-    case NDBEVENT::TE_ALTER:
-      if (pOp->tableNameChanged())
-      {  
-        DBUG_PRINT("info", ("Detected name change of table %s.%s",
-                            share->db, share->table_name));
-        /* do the rename of the table in the share */
-        share->table->s->db.str= share->db;
-        share->table->s->db.length= strlen(share->db);
-        share->table->s->table_name.str= share->table_name;
-        share->table->s->table_name.length= strlen(share->table_name);
-      }
-      ndb_handle_schema_change(thd, ndb, pOp, share);
-      break;
     case NDBEVENT::TE_CLUSTER_FAILURE:
     case NDBEVENT::TE_DROP:
       free_share(&schema_share);
       schema_share= 0;
+      // fall through
+    case NDBEVENT::TE_ALTER:
       ndb_handle_schema_change(thd, ndb, pOp, share);
       break;
     case NDBEVENT::TE_NODE_FAILURE:
@@ -1659,6 +1687,68 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
   DBUG_RETURN(0);
 }
 
+/*
+  process any operations that should be done after
+  the epoch is complete
+*/
+static void
+ndb_binlog_thread_handle_schema_event_post_epoch(THD *thd,
+                                                 List<Cluster_replication_schema>
+                                                 *post_epoch_log_list,
+                                                 List<Cluster_replication_schema>
+                                                 *post_epoch_unlock_list)
+{
+  DBUG_ENTER("ndb_binlog_thread_handle_schema_event_post_epoch");
+  Cluster_replication_schema *schema;
+  while ((schema= post_epoch_log_list->pop()))
+  {
+    DBUG_PRINT("info", ("log query_length: %d  query: '%s'",
+                        schema->query_length, schema->query));
+    {
+      char key[FN_REFLEN];
+      build_table_filename(key, sizeof(key), schema->db, schema->name, "");
+      NDB_SHARE *share= get_share(key, 0, false, false);
+      switch ((enum SCHEMA_OP_TYPE)schema->type)
+      {
+      case SOT_DROP_DB:
+      case SOT_DROP_TABLE:
+        break;
+      case SOT_RENAME_TABLE:
+      case SOT_ALTER_TABLE:
+        if (share && share->op)
+        {
+          break; /* discovery handled by binlog */
+        }
+        pthread_mutex_lock(&LOCK_open);
+        if (ndb_create_table_from_engine(thd, schema->db, schema->name))
+        {
+          sql_print_error("Could not discover table '%s.%s' from "
+                          "binlog schema event '%s' from node %d",
+                          schema->db, schema->name, schema->query,
+                          schema->node_id);
+        }
+        pthread_mutex_unlock(&LOCK_open);
+      default:
+        DBUG_ASSERT(false);
+      }
+      if (share)
+        free_share(&share);
+    }
+    {
+      char *thd_db_save= thd->db;
+      thd->db= schema->db;
+      thd->binlog_query(THD::STMT_QUERY_TYPE, schema->query,
+                        schema->query_length, FALSE,
+                        schema->name[0] == 0);
+      thd->db= thd_db_save;
+    }
+  }
+  while ((schema= post_epoch_unlock_list->pop()))
+  {
+    ndbcluster_update_slock(thd, schema->db, schema->name);
+  }      
+}
+
 /*
   Timer class for doing performance measurements
 */
@@ -2468,24 +2558,6 @@ ndb_binlog_thread_handle_non_data_event(Ndb *ndb, NdbEventOperation *pOp,
                         "op_old: %lx",
                        share->key, share, pOp, share->op, share->op_old));
     break;
-  case NDBEVENT::TE_ALTER:
-    if (pOp->tableNameChanged())
-    {
-      DBUG_PRINT("info", ("Detected name change of table %s.%s",
-                          share->db, share->table_name));
-      /* ToDo: remove printout */
-      if (ndb_extra_logging)
-        sql_print_information("NDB Binlog: rename table %s%s/%s -> %s.",
-                              share_prefix, share->table->s->db.str,
-                              share->table->s->table_name.str,
-                              share->key);
-      /* do the rename of the table in the share */
-      share->table->s->db.str= share->db;
-      share->table->s->db.length= strlen(share->db);
-      share->table->s->table_name.str= share->table_name;
-      share->table->s->table_name.length= strlen(share->table_name);
-    }
-    goto drop_alter_common;
   case NDBEVENT::TE_DROP:
     if (apply_status_share == share)
     {
@@ -2495,7 +2567,8 @@ ndb_binlog_thread_handle_non_data_event(Ndb *ndb, NdbEventOperation *pOp,
     /* ToDo: remove printout */
     if (ndb_extra_logging)
       sql_print_information("NDB Binlog: drop table %s.", share->key);
-drop_alter_common:
+    // fall through
+  case NDBEVENT::TE_ALTER:
     row.n_schemaops++;
     DBUG_PRINT("info", ("TABLE %s EVENT: %s  received share: 0x%lx  op: %lx  "
                         "share op: %lx  op_old: %lx",
@@ -3075,26 +3148,9 @@ pthread_handler_t ndb_binlog_thread_func(void *arg)
       }
     }
 
-    /*
-      process any operations that should be done after
-      the epoch is complete
-    */
-    {
-      Cluster_replication_schema *schema;
-      while ((schema= post_epoch_unlock_list.pop()))
-      {
-        ndbcluster_update_slock(thd, schema->db, schema->name);
-      }      
-      while ((schema= post_epoch_log_list.pop()))
-      {
-        char *thd_db_save= thd->db;
-        thd->db= schema->db;
-        thd->binlog_query(THD::STMT_QUERY_TYPE, schema->query,
-                          schema->query_length, FALSE,
-                          schema->name[0] == 0);
-        thd->db= thd_db_save;
-      }
-    }
+    ndb_binlog_thread_handle_schema_event_post_epoch(thd,
+                                                     &post_epoch_log_list,
+                                                     &post_epoch_unlock_list);
     free_root(&mem_root, MYF(0));
     *root_ptr= old_root;
     ndb_latest_handled_binlog_epoch= ndb_latest_received_binlog_epoch;