Fix for bug#16445 schema version error 241 after mysqld restart, always invalidate dictionary cache when receiving changed schema events drop table or alter table, regardless if done remotely or not (also includes cluster failures, but refreshing cache is a safe operation)

2 files changed, 60 insertions, 70 deletions

diff --git a/mysql-test/t/ndb_basic.test b/mysql-test/t/ndb_basic.test
index df94545abea..ab62e3dd13a 100644
--- a/mysql-test/t/ndb_basic.test
+++ b/mysql-test/t/ndb_basic.test
@@ -6,17 +6,6 @@ DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7;
 drop database if exists mysqltest;
 --enable_warnings
 
-# workaround for bug#16445
-# remove to reproduce bug and run tests drom ndb start
-# and with ndb_autodiscover disabled
-CREATE TABLE t1 (
-  pk1 INT NOT NULL PRIMARY KEY,
-  attr1 INT NOT NULL,
-  attr2 INT,
-  attr3 VARCHAR(10)
-) ENGINE=ndbcluster;
-drop table t1;
-
 #
 # Basic test to show that the NDB 
 # table handler is working
diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index ca3dcf605b0..1f416e1816d 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -1372,6 +1372,9 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
                          NDB_SHARE *share)
 {
   DBUG_ENTER("ndb_handle_schema_change");
+  TABLE* table= share->table;
+  TABLE_SHARE *table_share= table->s;
+  const char *dbname= table_share->db.str;
   bool do_close_cached_tables= FALSE;
   bool is_online_alter_table= FALSE;
   bool is_rename_table= FALSE;
@@ -1391,70 +1394,68 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
     }
   }
 
-  if (is_remote_change) /* includes CLUSTER_FAILURE */
+  /*
+    Refresh local dictionary cache by
+    invalidating table and all it's indexes
+  */
+  ndb->setDatabaseName(dbname);
+  Thd_ndb *thd_ndb= get_thd_ndb(thd);
+  DBUG_ASSERT(thd_ndb != NULL);
+  Ndb* old_ndb= thd_ndb->ndb;
+  thd_ndb->ndb= ndb;
+  ha_ndbcluster table_handler(table_share);
+  table_handler.set_dbname(share->key);
+  table_handler.set_tabname(share->key);
+  table_handler.open_indexes(ndb, table, TRUE);
+  table_handler.invalidate_dictionary_cache(TRUE);
+  thd_ndb->ndb= old_ndb;
+  
+  /*
+    Refresh local frm file and dictionary cache if
+    remote on-line alter table
+  */
+  if (is_remote_change && is_online_alter_table)
   {
-    TABLE* table= share->table;
-    TABLE_SHARE *table_share= table->s;
-    const char *dbname= table_share->db.str;
+    const char *tabname= table_share->table_name.str;
+    char key[FN_REFLEN];
+    const void *data= 0, *pack_data= 0;
+    uint length, pack_length;
+    int error;
+    NDBDICT *dict= ndb->getDictionary();
+    const NDBTAB *altered_table= pOp->getTable();
     
-    /* 
-       Invalidate table and all it's indexes
+    DBUG_PRINT("info", ("Detected frm change of table %s.%s",
+                        dbname, tabname));
+    build_table_filename(key, FN_LEN-1, dbname, tabname, NullS);
+    /*
+      If the frm of the altered table is different than the one on
+      disk then overwrite it with the new table definition
     */
-    ndb->setDatabaseName(dbname);
-    Thd_ndb *thd_ndb= get_thd_ndb(thd);
-    DBUG_ASSERT(thd_ndb != NULL);
-    Ndb* old_ndb= thd_ndb->ndb;
-    thd_ndb->ndb= ndb;
-    ha_ndbcluster table_handler(table_share);
-    table_handler.set_dbname(share->key);
-    table_handler.set_tabname(share->key);
-    table_handler.open_indexes(ndb, table, TRUE);
-    table_handler.invalidate_dictionary_cache(TRUE);
-    thd_ndb->ndb= old_ndb;
-    
-    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;
-      int error;
-      NDBDICT *dict= ndb->getDictionary();
-      const NDBTAB *altered_table= pOp->getTable();
-
-      DBUG_PRINT("info", ("Detected frm change of table %s.%s",
-                          dbname, tabname));
-      build_table_filename(key, FN_LEN-1, dbname, tabname, NullS);
-      /*
-        If the frm of the altered table is different than the one on
-        disk then overwrite it with the new table definition
-      */
-      if (readfrm(key, &data, &length) == 0 &&
-          packfrm(data, length, &pack_data, &pack_length) == 0 &&
-          cmp_frm(altered_table, pack_data, pack_length))
+    if (readfrm(key, &data, &length) == 0 &&
+        packfrm(data, length, &pack_data, &pack_length) == 0 &&
+        cmp_frm(altered_table, pack_data, pack_length))
+    {
+      DBUG_DUMP("frm", (char*)altered_table->getFrmData(), 
+                altered_table->getFrmLength());
+      pthread_mutex_lock(&LOCK_open);
+      const NDBTAB *old= dict->getTable(tabname);
+      if (!old &&
+          old->getObjectVersion() != altered_table->getObjectVersion())
+        dict->putTable(altered_table);
+      
+      if ((error= unpackfrm(&data, &length, altered_table->getFrmData())) ||
+          (error= writefrm(key, data, length)))
       {
-        DBUG_DUMP("frm", (char*)altered_table->getFrmData(), 
-                  altered_table->getFrmLength());
-        pthread_mutex_lock(&LOCK_open);
-        const NDBTAB *old= dict->getTable(tabname);
-        if (!old &&
-            old->getObjectVersion() != altered_table->getObjectVersion())
-          dict->putTable(altered_table);
-
-        if ((error= unpackfrm(&data, &length, altered_table->getFrmData())) ||
-            (error= writefrm(key, data, length)))
-        {
-          sql_print_information("NDB: Failed write frm for %s.%s, error %d",
-                                dbname, tabname, error);
-        }
-        ndbcluster_binlog_close_table(thd, share);
-        close_cached_tables((THD*) 0, 0, (TABLE_LIST*) 0, TRUE);
-        if ((error= ndbcluster_binlog_open_table(thd, share, 
-                                                 table_share, table)))
-          sql_print_information("NDB: Failed to re-open table %s.%s",
-                                dbname, tabname);
-        pthread_mutex_unlock(&LOCK_open);
+        sql_print_information("NDB: Failed write frm for %s.%s, error %d",
+                              dbname, tabname, error);
       }
+      ndbcluster_binlog_close_table(thd, share);
+      close_cached_tables((THD*) 0, 0, (TABLE_LIST*) 0, TRUE);
+      if ((error= ndbcluster_binlog_open_table(thd, share, 
+                                               table_share, table)))
+        sql_print_information("NDB: Failed to re-open table %s.%s",
+                              dbname, tabname);
+      pthread_mutex_unlock(&LOCK_open);
     }
   }