Merge ahristov@bk-internal.mysql.com:/home/bk/mysql-5.1-new

into lmy004.:/work/mysql-5.1-bug16406

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_binlog.cc

@@ -237,10 +237,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)
     {
@@ -315,21 +338,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)
@@ -648,11 +656,8 @@ static int ndbcluster_create_apply_status_table(THD *thd)
     if so, remove it since there is none in Ndb
   */
   {
-    strxnmov(buf, sizeof(buf),
-             mysql_data_home,
-             "/" NDB_REP_DB "/" NDB_APPLY_TABLE,
-             reg_ext, NullS);
-    unpack_filename(buf,buf);
+    build_table_filename(buf, sizeof(buf),
+                         NDB_REP_DB, NDB_APPLY_TABLE, reg_ext);
     my_delete(buf, MYF(0));
   }
 
@@ -700,11 +705,8 @@ static int ndbcluster_create_schema_table(THD *thd)
     if so, remove it since there is none in Ndb
   */
   {
-    strxnmov(buf, sizeof(buf),
-             mysql_data_home,
-             "/" NDB_REP_DB "/" NDB_SCHEMA_TABLE,
-             reg_ext, NullS);
-    unpack_filename(buf,buf);
+    build_table_filename(buf, sizeof(buf),
+                         NDB_REP_DB, NDB_SCHEMA_TABLE, reg_ext);
     my_delete(buf, MYF(0));
   }
 
@@ -929,8 +931,7 @@ int ndbcluster_log_schema_op(THD *thd, NDB_SHARE *share,
   if (get_a_share)
   {
     char key[FN_REFLEN];
-    (void)strxnmov(key, FN_REFLEN, share_prefix, db,
-                   "/", table_name, NullS);
+    build_table_filename(key, sizeof(key), db, table_name, "");
     share= get_share(key, 0, false, false);
   }
 
@@ -1358,6 +1359,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 *)
@@ -1375,21 +1377,20 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
         {
         case SOT_DROP_TABLE:
           /* binlog dropping table after any table operations */
+          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))
           {
@@ -1407,6 +1408,7 @@ 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 */
+          if (ndb_binlog_running)
             post_epoch_log_list->push_back(schema, mem_root);
           log_query= 0;
           break;
@@ -1422,8 +1424,8 @@ ndb_binlog_thread_handle_schema_event(THD *thd, Ndb *ndb,
         case SOT_CLEAR_SLOCK:
         {
           char key[FN_REFLEN];
-          (void)strxnmov(key, FN_REFLEN, share_prefix, schema->db,
-                         "/", schema->name, NullS);
+          build_table_filename(key, sizeof(key),
+                               schema->db, schema->name, "");
           NDB_SHARE *share= get_share(key, 0, false, false);
           if (share)
           {
@@ -1463,7 +1465,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;
@@ -1752,6 +1754,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));
@@ -1792,7 +1795,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);