Commit b1e20006 authored by unknown's avatar unknown

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


parent 38b1f7c7
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;
-- 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
...@@ -956,14 +956,14 @@ public: ...@@ -956,14 +956,14 @@ public:
/** /**
* Create defined table given defined Table instance * Create defined table given defined Table instance
* @param Table Table to create * @param Table instance to create
* @return 0 if successful otherwise -1. * @return 0 if successful otherwise -1.
*/ */
int createTable(const Table &); int createTable(const Table &);
/** /**
* Drop table given retrieved Table instance * Drop table given retrieved Table instance
* @param Table Table to drop * @param Table instance to drop
* @return 0 if successful otherwise -1. * @return 0 if successful otherwise -1.
*/ */
int dropTable(Table &); int dropTable(Table &);
...@@ -1027,7 +1027,15 @@ public: ...@@ -1027,7 +1027,15 @@ public:
*/ */
int dropIndex(const char * indexName, int dropIndex(const char * indexName,
const char * tableName); 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 * Get index with given name, NULL if undefined
* @param indexName Name of index to get. * @param indexName Name of index to get.
...@@ -1037,6 +1045,15 @@ public: ...@@ -1037,6 +1045,15 @@ public:
const Index * getIndex(const char * indexName, const Index * getIndex(const char * indexName,
const char * tableName); 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 #ifndef DOXYGEN_SHOULD_SKIP_INTERNAL
/** /**
* Invalidate cached index object * Invalidate cached index object
......
...@@ -786,6 +786,12 @@ NdbDictionary::Dictionary::dropIndex(const char * indexName, ...@@ -786,6 +786,12 @@ NdbDictionary::Dictionary::dropIndex(const char * indexName,
return m_impl.dropIndex(indexName, tableName); return m_impl.dropIndex(indexName, tableName);
} }
int
NdbDictionary::Dictionary::dropIndex(const Index & ind)
{
return m_impl.dropIndex(NdbIndexImpl::getImpl(ind));
}
const NdbDictionary::Index * const NdbDictionary::Index *
NdbDictionary::Dictionary::getIndex(const char * indexName, NdbDictionary::Dictionary::getIndex(const char * indexName,
const char * tableName) const char * tableName)
...@@ -796,6 +802,16 @@ NdbDictionary::Dictionary::getIndex(const char * indexName, ...@@ -796,6 +802,16 @@ NdbDictionary::Dictionary::getIndex(const char * indexName,
return 0; 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 void
NdbDictionary::Dictionary::invalidateIndex(const char * indexName, NdbDictionary::Dictionary::invalidateIndex(const char * indexName,
const char * tableName){ const char * tableName){
......
...@@ -2147,7 +2147,7 @@ NdbDictionaryImpl::dropIndex(const char * indexName, ...@@ -2147,7 +2147,7 @@ NdbDictionaryImpl::dropIndex(const char * indexName,
m_error.code = 4243; m_error.code = 4243;
return -1; 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 // If index stored in cache is incompatible with the one in the kernel
// we must clear the cache and try again // we must clear the cache and try again
if (ret == INCOMPATIBLE_VERSION) { if (ret == INCOMPATIBLE_VERSION) {
...@@ -2169,40 +2169,23 @@ NdbDictionaryImpl::dropIndex(const char * indexName, ...@@ -2169,40 +2169,23 @@ NdbDictionaryImpl::dropIndex(const char * indexName,
} }
int 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; NdbTableImpl * timpl = impl.m_table;
if (timpl == 0) { if (timpl == 0) {
m_error.code = 709; m_error.code = 709;
return -1; 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); int ret = m_receiver.dropIndex(impl, *timpl);
if(ret == 0){ if(ret == 0){
m_localHash.drop(internalIndexName); m_localHash.drop(timpl->m_internalName.c_str());
m_globalHash->lock(); m_globalHash->lock();
impl.m_table->m_status = NdbDictionary::Object::Invalid; timpl->m_status = NdbDictionary::Object::Invalid;
m_globalHash->drop(impl.m_table); m_globalHash->drop(timpl);
m_globalHash->unlock(); m_globalHash->unlock();
} }
return ret; return ret;
}
m_error.code = 4243;
return -1;
} }
int int
......
...@@ -379,7 +379,8 @@ public: ...@@ -379,7 +379,8 @@ public:
int createIndex(NdbIndexImpl &ix); int createIndex(NdbIndexImpl &ix);
int dropIndex(const char * indexName, int dropIndex(const char * indexName,
const char * tableName); const char * tableName);
int dropIndex(NdbIndexImpl &, const char * tableName); // int dropIndex(NdbIndexImpl &, const char * tableName);
int dropIndex(NdbIndexImpl &);
NdbTableImpl * getIndexTable(NdbIndexImpl * index, NdbTableImpl * getIndexTable(NdbIndexImpl * index,
NdbTableImpl * table); NdbTableImpl * table);
...@@ -397,6 +398,8 @@ public: ...@@ -397,6 +398,8 @@ public:
bool do_add_blob_tables); bool do_add_blob_tables);
NdbIndexImpl * getIndex(const char * indexName, NdbIndexImpl * getIndex(const char * indexName,
const char * tableName); const char * tableName);
NdbIndexImpl * getIndex(const char * indexName,
NdbTableImpl * table);
NdbIndexImpl * getIndexImpl(const char * name, const char * internalName); NdbIndexImpl * getIndexImpl(const char * name, const char * internalName);
NdbEventImpl * getEvent(const char * eventName); NdbEventImpl * getEvent(const char * eventName);
NdbEventImpl * getEventImpl(const char * internalName); NdbEventImpl * getEventImpl(const char * internalName);
...@@ -643,17 +646,24 @@ NdbDictionaryImpl::get_local_table_info(const char * internalTableName, ...@@ -643,17 +646,24 @@ NdbDictionaryImpl::get_local_table_info(const char * internalTableName,
return info; // autoincrement already initialized return info; // autoincrement already initialized
} }
inline inline
NdbIndexImpl * NdbIndexImpl *
NdbDictionaryImpl::getIndex(const char * indexName, NdbDictionaryImpl::getIndex(const char * indexName,
const char * tableName) 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; const char * internalIndexName = 0;
if (tableName) { if (table) {
NdbTableImpl * t = getTable(tableName); internalIndexName = m_ndb.internalizeIndexName(table, indexName);
if (t != 0)
internalIndexName = m_ndb.internalizeIndexName(t, indexName);
} else { } else {
internalIndexName = internalIndexName =
m_ndb.internalizeTableName(indexName); // Index is also a table m_ndb.internalizeTableName(indexName); // Index is also a table
......
...@@ -898,6 +898,8 @@ static int fix_unique_index_attr_order(NDB_INDEX_DATA &data, ...@@ -898,6 +898,8 @@ static int fix_unique_index_attr_order(NDB_INDEX_DATA &data,
DBUG_RETURN(0); DBUG_RETURN(0);
} }
int ha_ndbcluster::build_index_list(Ndb *ndb, TABLE *tab, enum ILBP phase) int ha_ndbcluster::build_index_list(Ndb *ndb, TABLE *tab, enum ILBP phase)
{ {
uint i; uint i;
...@@ -3926,7 +3928,6 @@ int ha_ndbcluster::create_index(const char *name, ...@@ -3926,7 +3928,6 @@ int ha_ndbcluster::create_index(const char *name,
DBUG_RETURN(0); DBUG_RETURN(0);
} }
/* /*
Rename a table in NDB Cluster Rename a table in NDB Cluster
*/ */
...@@ -3935,12 +3936,16 @@ int ha_ndbcluster::rename_table(const char *from, const char *to) ...@@ -3935,12 +3936,16 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
{ {
NDBDICT *dict; NDBDICT *dict;
char new_tabname[FN_HEADLEN]; char new_tabname[FN_HEADLEN];
char new_dbname[FN_HEADLEN];
const NDBTAB *orig_tab; const NDBTAB *orig_tab;
int result; int result;
bool recreate_indexes= FALSE;
NDBDICT::List index_list;
DBUG_ENTER("ha_ndbcluster::rename_table"); DBUG_ENTER("ha_ndbcluster::rename_table");
DBUG_PRINT("info", ("Renaming %s to %s", from, to)); DBUG_PRINT("info", ("Renaming %s to %s", from, to));
set_dbname(from); set_dbname(from);
set_dbname(to, new_dbname);
set_tabname(from); set_tabname(from);
set_tabname(to, new_tabname); set_tabname(to, new_tabname);
...@@ -3958,6 +3963,12 @@ int ha_ndbcluster::rename_table(const char *from, const char *to) ...@@ -3958,6 +3963,12 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
if (!(orig_tab= dict->getTable(m_tabname))) if (!(orig_tab= dict->getTable(m_tabname)))
ERR_RETURN(dict->getNdbError()); 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; m_table= (void *)orig_tab;
// Change current database to that of target table // Change current database to that of target table
set_dbname(to); set_dbname(to);
...@@ -3968,6 +3979,34 @@ int ha_ndbcluster::rename_table(const char *from, const char *to) ...@@ -3968,6 +3979,34 @@ int ha_ndbcluster::rename_table(const char *from, const char *to)
result= handler::rename_table(from, 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); DBUG_RETURN(result);
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment