Invalidate table including all it's indexes from binlog thread

sql/ha_ndbcluster.cc

@@ -519,6 +519,7 @@ void ha_ndbcluster::invalidate_dictionary_cache(bool global)
   {
     NDBINDEX *index = (NDBINDEX *) m_index[i].index;
     NDBINDEX *unique_index = (NDBINDEX *) m_index[i].unique_index;
+    if (!index && !unique_index) continue;
     NDB_INDEX_TYPE idx_type= m_index[i].type;
 
     switch (idx_type) {
@@ -1076,7 +1077,7 @@ int ha_ndbcluster::get_metadata(const char *path)
   m_table= (void *)tab; 
   m_table_info= NULL; // Set in external lock
   
-  DBUG_RETURN(open_indexes(ndb, table));
+  DBUG_RETURN(open_indexes(ndb, table, FALSE));
 }
 
 static int fix_unique_index_attr_order(NDB_INDEX_DATA &data,
@@ -1249,7 +1250,7 @@ int ha_ndbcluster::add_index_handle(THD *thd, NDBDICT *dict, KEY *key_info,
 /*
   Associate index handles for each index of a table
 */
-int ha_ndbcluster::open_indexes(Ndb *ndb, TABLE *tab)
+int ha_ndbcluster::open_indexes(Ndb *ndb, TABLE *tab, bool ignore_error)
 {
   uint i;
   int error= 0;
@@ -1263,6 +1264,9 @@ int ha_ndbcluster::open_indexes(Ndb *ndb, TABLE *tab)
   for (i= 0; i < tab->s->keys; i++, key_info++, key_name++)
   {
     if ((error= add_index_handle(thd, dict, key_info, *key_name, i)))
+      if (ignore_error)
+        m_index[i].index= m_index[i].unique_index= NULL;
+      else
         break;
   }
   
@@ -3739,7 +3743,7 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
       {
         m_table= (void *)tab;
         m_table_version = tab->getObjectVersion();
-        if (!(my_errno= open_indexes(ndb, table)))
+        if (!(my_errno= open_indexes(ndb, table, FALSE)))
           DBUG_RETURN(my_errno);
       }
       m_table_info= tab_info;

sql/ha_ndbcluster.h

@@ -691,6 +691,10 @@ static void set_tabname(const char *pathname, char *tabname);
 
 private:
   friend int ndbcluster_drop_database_impl(const char *path);
+  friend int ndb_handle_schema_change(THD *thd, 
+                                      Ndb *ndb, NdbEventOperation *pOp,
+                                      NDB_SHARE *share);
+
   int alter_table_name(const char *to);
   static int delete_table(ha_ndbcluster *h, Ndb *ndb,
 			  const char *path,
@@ -708,7 +712,7 @@ static void set_tabname(const char *pathname, char *tabname);
   int create_indexes(Ndb *ndb, TABLE *tab);
   void clear_index(int i);
   void clear_indexes();
-  int open_indexes(Ndb *ndb, TABLE *tab);
+  int open_indexes(Ndb *ndb, TABLE *tab, bool ignore_error);
   void renumber_indexes(Ndb *ndb, TABLE *tab);
   int drop_indexes(Ndb *ndb, TABLE *tab);
   int add_index_handle(THD *thd, NdbDictionary::Dictionary *dict,

sql/ha_ndbcluster_binlog.cc

@@ -1285,7 +1285,7 @@ ndbcluster_update_slock(THD *thd,
 /*
   Handle _non_ data events from the storage nodes
 */
-static int
+int
 ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
                          NDB_SHARE *share)
 {
@@ -1294,12 +1294,18 @@ ndb_handle_schema_change(THD *thd, Ndb *ndb, NdbEventOperation *pOp,
   if (pOp->getEventType() != NDBEVENT::TE_CLUSTER_FAILURE &&
       pOp->getReqNodeId() != g_ndb_cluster_connection->node_id())
   {
-    ndb->setDatabaseName(share->table->s->db.str);
-    ha_ndbcluster::invalidate_dictionary_cache(share->table->s,
-                                               ndb,
-                                               share->table->s->db.str,
-                                               share->table->s->table_name.str,
-                                               TRUE);
+    TABLE_SHARE *table_share= share->table->s;
+    TABLE* table= share->table;
+    
+    /* 
+       Invalidate table and all it's indexes
+    */
+    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);
+
     remote_drop_table= 1;
   }