Fix for Bug #8753 Invalid schema object version after dropping index

mysql-test/include/have_multi_ndb.inc

+# Setup connections to both MySQL Servers connected to the cluster
+connect (server1,127.0.0.1,root,,test,$MASTER_MYPORT,);
+connect (server2,127.0.0.1,root,,test,$MASTER_MYPORT1,);
+
+# Check that server1 has NDB  support
+connection server1;
+disable_query_log;
+--disable_warnings
+drop table if exists t1, t2;
+--enable_warnings
+flush tables;
+@r/have_ndb.require show variables like "have_ndbcluster";
+# @r/server_id.require show variables like "server_id";
+enable_query_log;
+
+# Check that server2 has NDB support
+connection server2;
+disable_query_log;
+--disable_warnings
+drop table if exists t1, t2;
+--enable_warnings
+flush tables;
+@r/have_ndb.require show variables like "have_ndbcluster";
+# @r/server_id1.require show variables like "server_id";
+enable_query_log;
+
+# Set the default connection to 'server1'
+connection server1;

mysql-test/r/ndb_alter_table.result

@@ -170,3 +170,17 @@ c
 4
 5
 drop table t1;
+create table t1 ( a int primary key, b varchar(10), c varchar(10), index (b) )
+engine=ndb;
+insert into t1 values (1,'one','one'), (2,'two','two'), (3,'three','three');
+create index c on t1(c);
+select * from t1 where b = 'two';
+a	b	c
+2	two	two
+alter table t1 drop index c;
+select * from t1 where b = 'two';
+ERROR 42S02: Table 'test.t1' doesn't exist
+select * from t1 where b = 'two';
+a	b	c
+2	two	two
+drop table t1;

mysql-test/r/ndb_multi.result

+drop table if exists t1, t2, t3, t4;
+flush status;
+create table t1 (a int) engine=ndbcluster;
+create table t2 (a int) engine=ndbcluster;
+insert into t1 value (2);
+insert into t2 value (3);
+select * from t1;
+a
+2
+select * from t2;
+a
+3
+show status like 'handler_discover%';
+Variable_name	Value
+Handler_discover	0
+flush status;
+select * from t1;
+a
+2
+update t1 set a=3 where a=2;
+show status like 'handler_discover%';
+Variable_name	Value
+Handler_discover	1
+create table t3 (a int not null primary key, b varchar(22),
+c int, last_col text) engine=ndb;
+insert into t3 values(1, 'Hi!', 89, 'Longtext column');
+create table t4 (pk int primary key, b int) engine=ndb;
+select * from t1;
+a
+3
+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
+t2
+t3
+t4
+drop table t1, t2, t3, t4;

mysql-test/t/ndb_alter_table.test

 -- source include/have_ndb.inc
+-- source include/have_multi_ndb.inc
 
 --disable_warnings
 DROP TABLE IF EXISTS t1;
@@ -133,6 +134,21 @@ INSERT INTO t1 VALUES (1,2,0),(18,19,4),(20,21,0);
 select c from t1 order by c;
 drop table t1;
 
+create table t1 ( a int primary key, b varchar(10), c varchar(10), index (b) )
+engine=ndb;
+insert into t1 values (1,'one','one'), (2,'two','two'), (3,'three','three');
+create index c on t1(c); 
+connection server2;
+select * from t1 where b = 'two';
+connection server1;
+alter table t1 drop index c;
+connection server2;
+--error 1146
+select * from t1 where b = 'two';
+select * from t1 where b = 'two';
+connection server1;
+drop table t1;
+
 #--disable_warnings
 #DROP TABLE IF EXISTS t2;
 #--enable_warnings

mysql-test/t/ndb_multi.test

+-- source include/have_ndb.inc
+-- source include/have_multi_ndb.inc
+
+
+--disable_warnings
+drop table if exists t1, t2, t3, t4;
+--enable_warnings
+
+flush status;
+
+# Create test tables on server1
+create table t1 (a int) engine=ndbcluster;
+create table t2 (a int) engine=ndbcluster;
+insert into t1 value (2);
+insert into t2 value (3);
+select * from t1;
+select * from t2;
+show status like 'handler_discover%';
+
+# Connect to server2 and use the tables from there
+connection server2;
+flush status;
+select * from t1;
+update t1 set a=3 where a=2;
+show status like 'handler_discover%';
+
+# Create a new table on server2
+create table t3 (a int not null primary key, b varchar(22),
+c int, last_col text) engine=ndb;
+insert into t3 values(1, 'Hi!', 89, 'Longtext column');
+create table t4 (pk int primary key, b int) engine=ndb;
+
+# Check that the tables are accessible from server1
+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;
+
+

sql/ha_ndbcluster.cc

@@ -328,6 +328,36 @@ void ha_ndbcluster::no_uncommitted_rows_reset(THD *thd)
     #   The mapped error code
 */
 
+void ha_ndbcluster::invalidateCache()
+{
+  NDBDICT *dict= get_ndb()->getDictionary();
+  DBUG_PRINT("info", ("invalidating %s", m_tabname));
+  dict->invalidateTable(m_tabname);
+  table->version=0L;			/* Free when thread is ready */
+  /* Invalidate indexes */
+  for (uint i= 0; i < table->keys; i++)
+  {
+    NDBINDEX *index = (NDBINDEX *) m_index[i].index;
+    NDBINDEX *unique_index = (NDBINDEX *) m_index[i].unique_index;
+    NDB_INDEX_TYPE idx_type= m_index[i].type;
+
+    
+    switch(m_index[i].type) {
+    case(PRIMARY_KEY_ORDERED_INDEX):
+    case(ORDERED_INDEX):
+      dict->invalidateIndex(index->getName(), m_tabname);
+      break;      
+    case(UNIQUE_ORDERED_INDEX):
+      dict->invalidateIndex(index->getName(), m_tabname);
+    case(UNIQUE_INDEX):
+      dict->invalidateIndex(unique_index->getName(), m_tabname);
+      break;
+    case(PRIMARY_KEY_INDEX):
+    case(UNDEFINED_INDEX):
+      break;
+    }
+  }
+}
 
 int ha_ndbcluster::ndb_err(NdbConnection *trans)
 {
@@ -339,11 +369,7 @@ int ha_ndbcluster::ndb_err(NdbConnection *trans)
   switch (err.classification) {
   case NdbError::SchemaError:
   {
-    Ndb *ndb= get_ndb();
-    NDBDICT *dict= ndb->getDictionary();
-    DBUG_PRINT("info", ("invalidateTable %s", m_tabname));
-    dict->invalidateTable(m_tabname);
-    table->version=0L;			/* Free when thread is ready */
+    invalidateCache();
     break;
   }
   default:
@@ -733,7 +759,7 @@ int ha_ndbcluster::get_metadata(const char *path)
       if (!invalidating_ndb_table)
       {
 	DBUG_PRINT("info", ("Invalidating table"));
-	dict->invalidateTable(m_tabname);
+        invalidateCache();
 	invalidating_ndb_table= TRUE;
       }
       else

sql/ha_ndbcluster.h

@@ -201,6 +201,7 @@ class ha_ndbcluster: public handler
   void print_results();
 
   longlong get_auto_increment();
+  void invalidateCache();
   int ndb_err(NdbConnection*);
   bool uses_blob_value(bool all_fields);