Bug#16997 Table rename that changes database does not rename indexes,...

Bug#16997  Table rename that changes database does not rename indexes, recreate indexes in new database

mysql-test/r/ndb_rename.result

+DROP TABLE IF EXISTS t1,t2;
+drop database if exists mysqltest;
+CREATE TABLE t1 (
+pk1 INT NOT NULL PRIMARY KEY,
+attr1 INT NOT NULL,
+attr2 INT,
+attr3 VARCHAR(10),
+INDEX i1(attr1)
+) ENGINE=ndbcluster;
+alter table t1 rename t2;
+create database ndbtest;
+alter table t2 rename ndbtest.t2;
+drop table ndbtest.t2;
+drop database ndbtest;

mysql-test/t/ndb_rename.test

+-- source include/have_ndb.inc
+-- source include/not_embedded.inc
+
+--disable_warnings
+DROP TABLE IF EXISTS t1,t2;
+drop database if exists mysqltest;
+--enable_warnings
+
+#
+# Table rename tests
+#
+
+#
+# Create a normal table with primary key
+#
+CREATE TABLE t1 (
+  pk1 INT NOT NULL PRIMARY KEY,
+  attr1 INT NOT NULL,
+  attr2 INT,
+  attr3 VARCHAR(10),
+  INDEX i1(attr1)
+) ENGINE=ndbcluster;
+
+alter table t1 rename t2;
+
+create database ndbtest;
+alter table t2 rename ndbtest.t2;
+
+drop table ndbtest.t2;
+drop database ndbtest;
+
+# End of 4.1 tests

ndb/include/ndbapi/NdbDictionary.hpp

@@ -956,14 +956,14 @@ public:
 
     /**
      * Create defined table given defined Table instance
-     * @param Table Table to create
+     * @param Table instance to create
      * @return 0 if successful otherwise -1.
      */
     int createTable(const Table &);
 
     /**
      * Drop table given retrieved Table instance
-     * @param Table Table to drop
+     * @param Table instance to drop
      * @return 0 if successful otherwise -1.
      */
     int dropTable(Table &);
@@ -1027,7 +1027,15 @@ public:
      */
     int dropIndex(const char * indexName,
 		  const char * tableName);
-    
+
+
+    /**
+     * Drop index the defined Index instance
+     * @param Index to drop
+     * @return 0 if successful otherwise -1.
+     */
+    int dropIndex(const Index &);    
+
     /**
      * Get index with given name, NULL if undefined
      * @param indexName  Name of index to get.
@@ -1037,6 +1045,15 @@ public:
     const Index * getIndex(const char * indexName,
 			   const char * tableName);
 
+    /**
+     * Get index with given name, NULL if undefined
+     * @param indexName  Name of index to get.
+     * @param Table instance table that index belongs to.
+     * @return  index if successful, otherwise 0.
+     */
+    const Index * getIndex(const char * indexName,
+			   const Table & table);
+
 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
     /**
      * Invalidate cached index object

ndb/src/ndbapi/NdbDictionary.cpp

@@ -786,6 +786,12 @@ NdbDictionary::Dictionary::dropIndex(const char * indexName,
   return m_impl.dropIndex(indexName, tableName);
 }
 
+int 
+NdbDictionary::Dictionary::dropIndex(const Index & ind)
+{
+  return m_impl.dropIndex(NdbIndexImpl::getImpl(ind));
+}
+
 const NdbDictionary::Index * 
 NdbDictionary::Dictionary::getIndex(const char * indexName,
 				    const char * tableName)
@@ -796,6 +802,16 @@ NdbDictionary::Dictionary::getIndex(const char * indexName,
   return 0;
 }
 
+const NdbDictionary::Index * 
+NdbDictionary::Dictionary::getIndex(const char * indexName,
+				    const Table & t)
+{
+  NdbIndexImpl * i = m_impl.getIndex(indexName, & NdbTableImpl::getImpl(t));
+  if(i)
+    return i->m_facade;
+  return 0;
+}
+
 void
 NdbDictionary::Dictionary::invalidateIndex(const char * indexName,
                                            const char * tableName){

ndb/src/ndbapi/NdbDictionaryImpl.cpp

@@ -2147,7 +2147,7 @@ NdbDictionaryImpl::dropIndex(const char * indexName,
     m_error.code = 4243;
     return -1;
   }
-  int ret = dropIndex(*idx, tableName);
+  int ret = dropIndex(*idx); //, tableName);
   // If index stored in cache is incompatible with the one in the kernel
   // we must clear the cache and try again
   if (ret == INCOMPATIBLE_VERSION) {
@@ -2169,40 +2169,23 @@ NdbDictionaryImpl::dropIndex(const char * indexName,
 }
 
 int
-NdbDictionaryImpl::dropIndex(NdbIndexImpl & impl, const char * tableName)
+NdbDictionaryImpl::dropIndex(NdbIndexImpl & impl)
 {
-  const char * indexName = impl.getName();
-  if (tableName || m_ndb.usingFullyQualifiedNames()) {
     NdbTableImpl * timpl = impl.m_table;
     
     if (timpl == 0) {
       m_error.code = 709;
       return -1;
     }
-
-    const char * internalIndexName = (tableName)
-      ?
-      m_ndb.internalizeIndexName(getTable(tableName), indexName)
-      :
-      m_ndb.internalizeTableName(indexName); // Index is also a table
-
-    if(impl.m_status == NdbDictionary::Object::New){
-      return dropIndex(indexName, tableName);
-    }
-    
     int ret = m_receiver.dropIndex(impl, *timpl);
     if(ret == 0){
-      m_localHash.drop(internalIndexName);
+      m_localHash.drop(timpl->m_internalName.c_str());
       m_globalHash->lock();
-      impl.m_table->m_status = NdbDictionary::Object::Invalid;
-      m_globalHash->drop(impl.m_table);
+      timpl->m_status = NdbDictionary::Object::Invalid;
+      m_globalHash->drop(timpl);
       m_globalHash->unlock();
     }
     return ret;
-  }
-
-  m_error.code = 4243;
-  return -1;
 }
 
 int

ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -379,7 +379,8 @@ public:
   int createIndex(NdbIndexImpl &ix);
   int dropIndex(const char * indexName, 
 		const char * tableName);
-  int dropIndex(NdbIndexImpl &, const char * tableName);
+  //  int dropIndex(NdbIndexImpl &, const char * tableName);
+  int dropIndex(NdbIndexImpl &);
   NdbTableImpl * getIndexTable(NdbIndexImpl * index, 
 			       NdbTableImpl * table);
 
@@ -397,6 +398,8 @@ public:
 					      bool do_add_blob_tables);
   NdbIndexImpl * getIndex(const char * indexName,
 			  const char * tableName);
+  NdbIndexImpl * getIndex(const char * indexName,
+			  NdbTableImpl * table);
   NdbIndexImpl * getIndexImpl(const char * name, const char * internalName);
   NdbEventImpl * getEvent(const char * eventName);
   NdbEventImpl * getEventImpl(const char * internalName);
@@ -643,17 +646,24 @@ NdbDictionaryImpl::get_local_table_info(const char * internalTableName,
   return info; // autoincrement already initialized
 }
 
+
 inline
 NdbIndexImpl * 
 NdbDictionaryImpl::getIndex(const char * indexName,
 			    const char * tableName)
 {
-  if (tableName || m_ndb.usingFullyQualifiedNames()) {
+  return getIndex(indexName, (tableName) ? getTable(tableName) : NULL);
+}
+
+inline
+NdbIndexImpl * 
+NdbDictionaryImpl::getIndex(const char * indexName,
+			    NdbTableImpl * table)
+{
+  if (table || m_ndb.usingFullyQualifiedNames()) {
     const char * internalIndexName = 0;
-    if (tableName) {
-      NdbTableImpl * t = getTable(tableName);
-      if (t != 0)
-        internalIndexName = m_ndb.internalizeIndexName(t, indexName);
+    if (table) {
+      internalIndexName = m_ndb.internalizeIndexName(table, indexName);
     } else {
       internalIndexName =
 	m_ndb.internalizeTableName(indexName); // Index is also a table

sql/ha_ndbcluster.cc

@@ -898,6 +898,8 @@ static int fix_unique_index_attr_order(NDB_INDEX_DATA &data,
   DBUG_RETURN(0);
 }
 
+
+
 int ha_ndbcluster::build_index_list(Ndb *ndb, TABLE *tab, enum ILBP phase)
 {
   uint i;
@@ -3926,7 +3928,6 @@ int ha_ndbcluster::create_index(const char *name,
   DBUG_RETURN(0);  
 }
 
-
 /*
   Rename a table in NDB Cluster
 */
@@ -3935,12 +3936,16 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
 {
   NDBDICT *dict;
   char new_tabname[FN_HEADLEN];
+  char new_dbname[FN_HEADLEN];
   const NDBTAB *orig_tab;
   int result;
+  bool recreate_indexes= FALSE;
+  NDBDICT::List index_list;
 
   DBUG_ENTER("ha_ndbcluster::rename_table");
   DBUG_PRINT("info", ("Renaming %s to %s", from, to));
   set_dbname(from);
+  set_dbname(to, new_dbname);
   set_tabname(from);
   set_tabname(to, new_tabname);
 
@@ -3958,6 +3963,12 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
     if (!(orig_tab= dict->getTable(m_tabname)))
       ERR_RETURN(dict->getNdbError());
   }
+  if (my_strcasecmp(system_charset_info, new_dbname, m_dbname))
+  {
+    dict->listIndexes(index_list, m_tabname);
+    recreate_indexes= TRUE;
+  }
+
   m_table= (void *)orig_tab;
   // Change current database to that of target table
   set_dbname(to);
@@ -3968,6 +3979,34 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
     result= handler::rename_table(from, to);
   }
 
+  // If we are moving tables between databases, we need to recreate
+  // indexes
+  if (recreate_indexes)
+  {
+    const NDBTAB *new_tab;
+    set_tabname(to);
+    if (!(new_tab= dict->getTable(m_tabname)))
+      ERR_RETURN(dict->getNdbError());
+
+    for (unsigned i = 0; i < index_list.count; i++) {
+        NDBDICT::List::Element& index_el = index_list.elements[i];
+	set_dbname(from);
+	ndb->setDatabaseName(m_dbname);
+	const NDBINDEX * index= dict->getIndex(index_el.name,  *new_tab);
+	set_dbname(to);
+	ndb->setDatabaseName(m_dbname);
+	DBUG_PRINT("info", ("Creating index %s/%s", 
+			    m_dbname, index->getName()));
+	dict->createIndex(*index);
+        DBUG_PRINT("info", ("Dropping index %s/%s", 
+			    m_dbname, index->getName()));
+	
+	set_dbname(from);
+	ndb->setDatabaseName(m_dbname);
+	dict->dropIndex(*index);
+    }
+  }
+
   DBUG_RETURN(result);
 }