Bug #17414 ndb schema distribution functionality does not work on mysql servers without binlog

mysql-test/r/ndb_alter_table_stm.result

@@ -8,8 +8,6 @@ a	b	c
 2	two	two
 alter table t1 drop index c;
 select * from t1 where c = 'two';
-ERROR HY000: Table definition has changed, please retry transaction
-select * from t1 where c = 'two';
 a	b	c
 2	two	two
 drop table t1;

mysql-test/r/ndb_multi.result

@@ -30,14 +30,6 @@ drop table t1;
 create table t1 (a int) engine=ndbcluster;
 insert into t1 value (2);
 select * from t1;
-ERROR HY000: Table definition has changed, please retry transaction
-show warnings;
-Level	Code	Message
-Error	1296	Got error 241 'Invalid schema object version' from NDB
-Error	1412	Table definition has changed, please retry transaction
-Error	1105	Unknown error
-flush table t1;
-select * from t1;
 a
 2
 flush status;
@@ -58,15 +50,9 @@ a
 select * from t3;
 a	b	c	last_col
 1	Hi!	89	Longtext column
-show status like 'handler_discover%';
-Variable_name	Value
-Handler_discover	1
 show tables like 't4';
 Tables_in_test (t4)
 t4
-show status like 'handler_discover%';
-Variable_name	Value
-Handler_discover	2
 show tables;
 Tables_in_test
 t1
@@ -74,4 +60,3 @@ t2
 t3
 t4
 drop table t1, t2, t3, t4;
-drop table t1, t3, t4;

mysql-test/t/ndb_alter_table_stm.test

@@ -17,8 +17,6 @@ select * from t1 where c = 'two';
 connection server1;
 alter table t1 drop index c;
 connection server2;
--- error 1412
-select * from t1 where c = 'two';
 select * from t1 where c = 'two';
 connection server1;
 drop table t1;

mysql-test/t/ndb_multi.test

@@ -41,11 +41,12 @@ drop table t1;
 create table t1 (a int) engine=ndbcluster;
 insert into t1 value (2);
 connection server1;
-# Currently a retry is required remotely
---error 1412
-select * from t1;
-show warnings;
-flush table t1;
+## Currently a retry is required remotely
+#--error 1412
+#select * from t1;
+#show warnings;
+#flush table t1;
+# Table definition change should be propagated automatically
 select * from t1;
 
 # Connect to server2 and use the tables from there
@@ -65,13 +66,9 @@ create table t4 (pk int primary key, b int) engine=ndb;
 connection server1;
 select * from t1;
 select * from t3;
-show status like 'handler_discover%';
 show tables like 't4';
-show status like 'handler_discover%';
 show tables;
 
 drop table t1, t2, t3, t4;
-connection server2;
-drop table t1, t3, t4;
 
 # End of 4.1 tests

sql/ha_ndbcluster.cc

@@ -34,6 +34,7 @@
 #include <ndbapi/NdbIndexStat.hpp>
 
 #include "ha_ndbcluster_binlog.h"
+#include "ha_ndbcluster_tables.h"
 
 #ifdef ndb_dynamite
 #undef assert
@@ -4381,6 +4382,12 @@ int ha_ndbcluster::create(const char *name,
       const NDBTAB *t= dict->getTable(m_tabname);
       String event_name(INJECTOR_EVENT_LEN);
       ndb_rep_event_name(&event_name,m_dbname,m_tabname);
+      int do_event_op= ndb_binlog_running;
+
+      if (!schema_share &&
+          strcmp(share->db, NDB_REP_DB) == 0 &&
+          strcmp(share->table_name, NDB_SCHEMA_TABLE) == 0)
+        do_event_op= 1;
 
       /*
         Always create an event for the table, as other mysql servers
@@ -4389,7 +4396,7 @@ int ha_ndbcluster::create(const char *name,
       if (ndbcluster_create_event(ndb, t, event_name.c_ptr(), share) < 0)
       {
         /* this is only a serious error if the binlog is on */
-	if (share && ndb_binlog_running)
+	if (share && do_event_op)
 	{
           push_warning_printf(current_thd, MYSQL_ERROR::WARN_LEVEL_ERROR,
                               ER_GET_ERRMSG, ER(ER_GET_ERRMSG),
@@ -4402,14 +4409,14 @@ int ha_ndbcluster::create(const char *name,
         sql_print_information("NDB Binlog: CREATE TABLE Event: %s",
                               event_name.c_ptr());
 
-      if (share && ndb_binlog_running &&
+      if (share && do_event_op &&
           ndbcluster_create_event_ops(share, t, event_name.c_ptr()) < 0)
       {
         sql_print_error("NDB Binlog: FAILED CREATE TABLE event operations."
                         " Event: %s", name2);
         /* a warning has been issued to the client */
       }
-      if (share && !ndb_binlog_running)
+      if (share && !do_event_op)
         share->flags|= NSF_NO_BINLOG;
       ndbcluster_log_schema_op(current_thd, share,
                                current_thd->query, current_thd->query_length,
@@ -4692,9 +4699,8 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
       ERR_RETURN(dict->getNdbError());
   }
 #ifdef HAVE_NDB_BINLOG
-  NDB_SHARE *share= 0;
-  if (ndb_binlog_running &&
-      (share= get_share(from, 0, false)))
+  NDB_SHARE *share= get_share(from, 0, false);
+  if (share)
   {
     int r= rename_share(share, to);
     DBUG_ASSERT(r == 0);
@@ -4755,7 +4761,7 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
       if (ndb_extra_logging)
         sql_print_information("NDB Binlog: RENAME Event: %s",
                               event_name.c_ptr());
-      if (share)
+      if (share && ndb_binlog_running)
       {
         if (ndbcluster_create_event_ops(share, ndbtab,
                                         event_name.c_ptr()) < 0)
@@ -6615,16 +6621,19 @@ static int rename_share(NDB_SHARE *share, const char *new_key)
              ("db.tablename: %s.%s  use_count: %d  commit_count: %d",
               share->db, share->table_name,
               share->use_count, share->commit_count));
-  DBUG_PRINT("rename_share",
-             ("table->s->db.table_name: %s.%s",
-              share->table->s->db.str, share->table->s->table_name.str));
-
-  if (share->op == 0)
+  if (share->table)
   {
-    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_PRINT("rename_share",
+               ("table->s->db.table_name: %s.%s",
+                share->table->s->db.str, share->table->s->table_name.str));
+
+    if (share->op == 0)
+    {
+      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);
+    }
   }
   /* else rename will be handled when the ALTER event comes */
   share->old_names= old_key;

sql/ha_ndbcluster_binlog.cc

@@ -240,10 +240,33 @@ void ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *_table)
 {
   THD *thd= current_thd;
   MEM_ROOT *mem_root= &share->mem_root;
+  int do_event_op= ndb_binlog_running;
 
   share->op= 0;
   share->table= 0;
-  if (!ndb_binlog_running)
+
+  if (!schema_share &&
+      strcmp(share->db, NDB_REP_DB) == 0 &&
+      strcmp(share->table_name, NDB_SCHEMA_TABLE) == 0)
+    do_event_op= 1;
+
+  {
+    int i, no_nodes= g_ndb_cluster_connection->no_db_nodes();
+    share->subscriber_bitmap= (MY_BITMAP*)
+      alloc_root(mem_root, no_nodes * sizeof(MY_BITMAP));
+    for (i= 0; i < no_nodes; i++)
+    {
+      bitmap_init(&share->subscriber_bitmap[i],
+                  (Uint32*)alloc_root(mem_root, max_ndb_nodes/8),
+                  max_ndb_nodes, false);
+      bitmap_clear_all(&share->subscriber_bitmap[i]);
+    }
+    bitmap_init(&share->slock_bitmap, share->slock,
+                sizeof(share->slock)*8, false);
+    bitmap_clear_all(&share->slock_bitmap);
+  }
+
+  if (!do_event_op)
   {
     if (_table)
     {
@@ -318,21 +341,6 @@ void ndbcluster_binlog_init_share(NDB_SHARE *share, TABLE *_table)
     share->ndb_value[1]= (NdbValue*)
       alloc_root(mem_root, sizeof(NdbValue) * table->s->fields
                  +1 /*extra for hidden key*/);
-    {
-      int i, no_nodes= g_ndb_cluster_connection->no_db_nodes();
-      share->subscriber_bitmap= (MY_BITMAP*)
-        alloc_root(mem_root, no_nodes * sizeof(MY_BITMAP));
-      for (i= 0; i < no_nodes; i++)
-      {
-        bitmap_init(&share->subscriber_bitmap[i],
-                    (Uint32*)alloc_root(mem_root, max_ndb_nodes/8),
-                    max_ndb_nodes, false);
-        bitmap_clear_all(&share->subscriber_bitmap[i]);
-      }
-      bitmap_init(&share->slock_bitmap, share->slock,
-                  sizeof(share->slock)*8, false);
-      bitmap_clear_all(&share->slock_bitmap);
-    }
     if (table->s->primary_key == MAX_KEY)
       share->flags|= NSF_HIDDEN_PK;
     if (table->s->blob_fields != 0)
@@ -1361,6 +1369,7 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
     switch (ev_type)
     {
     case NDBEVENT::TE_UPDATE:
+      /* fall through */
     case NDBEVENT::TE_INSERT:
     {
       Cluster_replication_schema *schema= (Cluster_replication_schema *)
@@ -1378,21 +1387,20 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
         {
         case SOT_DROP_TABLE:
           /* binlog dropping table after any table operations */
-          post_epoch_log_list->push_back(schema, mem_root);
+          if (ndb_binlog_running)
+            post_epoch_log_list->push_back(schema, mem_root);
           log_query= 0;
           break;
         case SOT_RENAME_TABLE:
           /* fall through */
         case SOT_ALTER_TABLE:
-          /* fall through */
-          if (!ndb_binlog_running)
+          if (ndb_binlog_running)
           {
             log_query= 1;
             break; /* discovery will be handled by binlog */
           }
           /* fall through */
         case SOT_CREATE_TABLE:
-          /* fall through */
           pthread_mutex_lock(&LOCK_open);
           if (ndb_create_table_from_engine(thd, schema->db, schema->name))
           {
@@ -1410,7 +1418,8 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
                     TRUE,    /* print error */
                     TRUE);   /* don't binlog the query */
           /* binlog dropping database after any table operations */
-          post_epoch_log_list->push_back(schema, mem_root);
+          if (ndb_binlog_running)
+            post_epoch_log_list->push_back(schema, mem_root);
           log_query= 0;
           break;
         case SOT_CREATE_DB:
@@ -1466,7 +1475,7 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
           }
         }
 
-        if (log_query)
+        if (log_query && ndb_binlog_running)
         {
           char *thd_db_save= thd->db;
           thd->db= schema->db;
@@ -1755,6 +1764,7 @@ int ndbcluster_create_binlog_setup(Ndb *ndb, const char *key,
                                    const char *table_name,
                                    my_bool share_may_exist)
 {
+  int do_event_op= ndb_binlog_running;
   DBUG_ENTER("ndbcluster_create_binlog_setup");
   DBUG_PRINT("enter",("key: %s  key_len: %d  %s.%s  share_may_exist: %d",
                       key, key_len, db, table_name, share_may_exist));
@@ -1795,7 +1805,12 @@ int ndbcluster_create_binlog_setup(Ndb *ndb, const char *key,
                     "allocating table share for %s failed", key);
   }
 
-  if (!ndb_binlog_running)
+  if (!schema_share &&
+      strcmp(share->db, NDB_REP_DB) == 0 &&
+      strcmp(share->table_name, NDB_SCHEMA_TABLE) == 0)
+    do_event_op= 1;
+
+  if (!do_event_op)
   {
     share->flags|= NSF_NO_BINLOG;
     pthread_mutex_unlock(&ndbcluster_mutex);