Fixed incorrectly use if table with wrong schema version

mysql-test/r/ndb_alter_table.result

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

mysql-test/t/ndb_alter_table.test

@@ -147,6 +147,8 @@ select * from t1 where b = 'two';
 connection server1;
 alter table t1 drop index c;
 connection server2;
+# This should fail since index information is not automatically refreshed
+--error 1105
 select * from t1 where b = 'two';
 select * from t1 where b = 'two';
 connection server1;

sql/ha_ndbcluster.cc

@@ -50,6 +50,8 @@ static const char *ha_ndb_ext=".ndb";
 #define NDB_FAILED_AUTO_INCREMENT ~(Uint64)0
 #define NDB_AUTO_INCREMENT_RETRIES 10
 
+#define NDB_INVALID_SCHEMA_OBJECT 241
+
 #define ERR_PRINT(err) \
   DBUG_PRINT("error", ("%d  message: %s", err.code, err.message))
 
@@ -212,7 +214,21 @@ Thd_ndb::Thd_ndb()
 Thd_ndb::~Thd_ndb()
 {
   if (ndb)
+  {
+#ifndef DBUG_OFF
+    Ndb::Free_list_usage tmp; tmp.m_name= 0;
+    while (ndb->get_free_list_usage(&tmp))
+    {
+      uint leaked= (uint) tmp.m_created - tmp.m_free;
+      if (leaked)
+        fprintf(stderr, "NDB: Found %u %s%s that %s not been released\n",
+                leaked, tmp.m_name,
+                (leaked == 1)?"":"'s",
+                (leaked == 1)?"has":"have");
+    }
+#endif
     delete ndb;
+  }
   ndb= 0;
 }
 
@@ -3269,15 +3285,19 @@ int ha_ndbcluster::external_lock(THD *thd, int lock_type)
         DBUG_PRINT("info", ("Table schema version: %d", 
                             tab->getObjectVersion()));
       }
-      if (m_table != (void *)tab || m_table_version < tab->getObjectVersion())
+      if (m_table != (void *)tab)
       {
-        /*
-          The table has been altered, refresh the index list
-        */
-        build_index_list(ndb, table, ILBP_OPEN);  
         m_table= (void *)tab;
         m_table_version = tab->getObjectVersion();
       }
+      else if (m_table_version < tab->getObjectVersion())
+      {
+        /*
+          The table has been altered, caller has to retry
+        */
+        NdbError err= ndb->getNdbError(NDB_INVALID_SCHEMA_OBJECT);
+        DBUG_RETURN(ndb_to_mysql_error(&err));
+      }
       m_table_info= tab_info;
     }
     no_uncommitted_rows_init(thd);
@@ -4657,7 +4677,21 @@ bool ndbcluster_end()
 {
   DBUG_ENTER("ndbcluster_end");
   if(g_ndb)
+  {
+#ifndef DBUG_OFF
+    Ndb::Free_list_usage tmp; tmp.m_name= 0;
+    while (g_ndb->get_free_list_usage(&tmp))
+    {
+      uint leaked= (uint) tmp.m_created - tmp.m_free;
+      if (leaked)
+        fprintf(stderr, "NDB: Found %u %s%s that %s not been released\n",
+                leaked, tmp.m_name,
+                (leaked == 1)?"":"'s",
+                (leaked == 1)?"has":"have");
+    }
+#endif
     delete g_ndb;
+  }
   g_ndb= NULL;
   if (g_ndb_cluster_connection)
     delete g_ndb_cluster_connection;