ndb - bug#20053

  make sure we can only drop files from correct file group

mysql-test/r/ndb_dd_ddl.result

@@ -188,6 +188,19 @@ ENGINE NDB;
 CREATE INDEX b_i on t1(b);
 CREATE INDEX bc_i on t1(b, c);
 DROP TABLE t1;
+CREATE TABLESPACE ts2
+ADD DATAFILE 'datafile3.dat'
+USE LOGFILE GROUP lg1
+INITIAL_SIZE 1M
+ENGINE NDB;
+ALTER TABLESPACE ts1
+DROP DATAFILE 'datafile3.dat'
+ENGINE NDB;
+ERROR HY000: Failed to alter:  NO SUCH FILE
+ALTER TABLESPACE ts2
+DROP DATAFILE 'datafile2.dat'
+ENGINE NDB;
+ERROR HY000: Failed to alter:  NO SUCH FILE
 ALTER TABLESPACE ts1
 DROP DATAFILE 'datafile2.dat'
 ENGINE NDB;
@@ -196,6 +209,11 @@ DROP DATAFILE 'datafile.dat'
 ENGINE NDB;
 DROP TABLESPACE ts1
 ENGINE NDB;
+ALTER TABLESPACE ts2
+DROP DATAFILE 'datafile3.dat'
+ENGINE NDB;
+DROP TABLESPACE ts2
+ENGINE NDB;
 DROP LOGFILE GROUP lg1
 ENGINE NDB;
 **** End = And No =  ****

mysql-test/t/ndb_dd_ddl.test

@@ -280,6 +280,25 @@ CREATE INDEX bc_i on t1(b, c);
 
 DROP TABLE t1;
 
+# bug#20053
+
+CREATE TABLESPACE ts2
+ADD DATAFILE 'datafile3.dat'
+USE LOGFILE GROUP lg1
+INITIAL_SIZE 1M
+ENGINE NDB;
+
+--error ER_ALTER_FILEGROUP_FAILED
+ALTER TABLESPACE ts1
+DROP DATAFILE 'datafile3.dat'
+ENGINE NDB;
+
+--error ER_ALTER_FILEGROUP_FAILED
+ALTER TABLESPACE ts2
+DROP DATAFILE 'datafile2.dat'
+ENGINE NDB;
+# bug#20053
+
 ALTER TABLESPACE ts1
 DROP DATAFILE 'datafile2.dat'
 ENGINE NDB;
@@ -291,6 +310,13 @@ ENGINE NDB;
 DROP TABLESPACE ts1
 ENGINE NDB;
 
+ALTER TABLESPACE ts2
+DROP DATAFILE 'datafile3.dat'
+ENGINE NDB;
+
+DROP TABLESPACE ts2
+ENGINE NDB;
+
 DROP LOGFILE GROUP lg1
 ENGINE NDB;
 

sql/ha_ndbcluster.cc

@@ -9949,7 +9949,7 @@ int ndbcluster_alter_tablespace(THD* thd, st_alter_tablespace *info)
   }
 
   NdbError err;
-  NDBDICT *dict = ndb->getDictionary();
+  NDBDICT *dict= ndb->getDictionary();
   int error;
   const char * errmsg;
   LINT_INIT(errmsg);
@@ -10013,9 +10013,12 @@ int ndbcluster_alter_tablespace(THD* thd, st_alter_tablespace *info)
     }
     else if(info->ts_alter_tablespace_type == ALTER_TABLESPACE_DROP_FILE)
     {
-      NdbDictionary::Datafile df = dict->getDatafile(0, 
-						     info->data_file_name);
-      if (strcmp(df.getPath(), info->data_file_name) == 0)
+      NdbDictionary::Tablespace ts= dict->getTablespace(info->tablespace_name);
+      NdbDictionary::Datafile df= dict->getDatafile(0, info->data_file_name);
+      NdbDictionary::ObjectId objid;
+      df.getTablespaceId(&objid);
+      if (ts.getObjectId() == objid.getObjectId() && 
+	  strcmp(df.getPath(), info->data_file_name) == 0)
       {
 	errmsg= " DROP DATAFILE";
 	if (dict->dropDatafile(df))
@@ -10344,10 +10347,12 @@ static int ndbcluster_fill_files_table(THD *thd, TABLE_LIST *tables,
       table->field[c++]->set_null(); // TABLE_NAME
 
       // LOGFILE_GROUP_NAME
+      NdbDictionary::ObjectId objid;
+      uf.getLogfileGroupId(&objid);
       table->field[c++]->store(uf.getLogfileGroup(),
                                strlen(uf.getLogfileGroup()),
                                system_charset_info);
-      table->field[c++]->store(uf.getLogfileGroupId()); // LOGFILE_GROUP_NUMBER
+      table->field[c++]->store(objid.getObjectId()); // LOGFILE_GROUP_NUMBER
       table->field[c++]->store(ndbcluster_hton_name,
                                ndbcluster_hton_name_length,
                                system_charset_info); // ENGINE

storage/ndb/include/ndbapi/NdbDictionary.hpp

@@ -184,7 +184,7 @@ public:
     virtual int getObjectId() const;
     
   private:
-    friend class Dictionary;
+    friend class NdbDictObjectImpl;
     class NdbDictObjectImpl & m_impl;
   };
   
@@ -1466,7 +1466,7 @@ public:
     void setTablespace(const char * name);
     void setTablespace(const class Tablespace &);
     const char * getTablespace() const;
-    Uint32 getTablespaceId() const;
+    void getTablespaceId(ObjectId * dst) const;
 
     void setNode(Uint32 nodeId);
     Uint32 getNode() const;
@@ -1509,7 +1509,7 @@ public:
     void setLogfileGroup(const char * name);
     void setLogfileGroup(const class LogfileGroup &);
     const char * getLogfileGroup() const;
-    Uint32 getLogfileGroupId() const;
+    void getLogfileGroupId(ObjectId * dst) const;
 
     void setNode(Uint32 nodeId);
     Uint32 getNode() const;

storage/ndb/src/ndbapi/NdbDictionary.cpp

@@ -1224,9 +1224,14 @@ NdbDictionary::Datafile::getTablespace() const {
   return m_impl.m_filegroup_name.c_str();
 }
 
-Uint32
-NdbDictionary::Datafile::getTablespaceId() const {
-  return m_impl.m_filegroup_id;
+void
+NdbDictionary::Datafile::getTablespaceId(NdbDictionary::ObjectId* dst) const 
+{
+  if (dst)
+  {
+    NdbDictObjectImpl::getImpl(* dst).m_id = m_impl.m_filegroup_id;
+    NdbDictObjectImpl::getImpl(* dst).m_version = m_impl.m_filegroup_version;
+  }
 }
 
 NdbDictionary::Object::Status
@@ -1310,9 +1315,14 @@ NdbDictionary::Undofile::getLogfileGroup() const {
   return m_impl.m_filegroup_name.c_str();
 }
 
-Uint32
-NdbDictionary::Undofile::getLogfileGroupId() const {
-  return m_impl.m_filegroup_id;
+void
+NdbDictionary::Undofile::getLogfileGroupId(NdbDictionary::ObjectId * dst)const 
+{
+  if (dst)
+  {
+    NdbDictObjectImpl::getImpl(* dst).m_id = m_impl.m_filegroup_id;
+    NdbDictObjectImpl::getImpl(* dst).m_version = m_impl.m_filegroup_version;
+  }
 }
 
 NdbDictionary::Object::Status
@@ -1829,7 +1839,8 @@ NdbDictionary::Dictionary::createLogfileGroup(const LogfileGroup & lg,
 					      ObjectId * obj)
 {
   return m_impl.createLogfileGroup(NdbLogfileGroupImpl::getImpl(lg),
-				   obj ? &obj->m_impl : 0);
+				   obj ? 
+				   & NdbDictObjectImpl::getImpl(* obj) : 0);
 }
 
 int
@@ -1852,7 +1863,8 @@ NdbDictionary::Dictionary::createTablespace(const Tablespace & lg,
 					    ObjectId * obj)
 {
   return m_impl.createTablespace(NdbTablespaceImpl::getImpl(lg),
-				 obj ? &obj->m_impl : 0);
+				 obj ? 
+				 & NdbDictObjectImpl::getImpl(* obj) : 0);
 }
 
 int
@@ -1887,7 +1899,7 @@ NdbDictionary::Dictionary::createDatafile(const Datafile & df,
 {
   return m_impl.createDatafile(NdbDatafileImpl::getImpl(df), 
 			       force,
-			       obj ? &obj->m_impl : 0);
+			       obj ? & NdbDictObjectImpl::getImpl(* obj) : 0);
 }
 
 int
@@ -1913,7 +1925,7 @@ NdbDictionary::Dictionary::createUndofile(const Undofile & df,
 {
   return m_impl.createUndofile(NdbUndofileImpl::getImpl(df), 
 			       force,
-			       obj ? &obj->m_impl : 0);
+			       obj ? & NdbDictObjectImpl::getImpl(* obj) : 0);
 }
 
 int

storage/ndb/src/ndbapi/NdbDictionaryImpl.hpp

@@ -46,14 +46,22 @@ public:
   NdbDictionary::Object::Status m_status;
   
   bool change();
+  
+  static NdbDictObjectImpl & getImpl(NdbDictionary::ObjectId & t) { 
+    return t.m_impl;
+  }
+  static const NdbDictObjectImpl & getImpl(const NdbDictionary::ObjectId & t){
+    return t.m_impl;
+  }
+  
 protected:
+  friend class NdbDictionary::ObjectId;
+
   NdbDictObjectImpl(NdbDictionary::Object::Type type) :
     m_type(type),
     m_status(NdbDictionary::Object::New) {
     m_id = -1;
   }
-  
-  friend class NdbDictionary::ObjectId;
 };
 
 /**

storage/ndb/tools/restore/consumer_restore.cpp

@@ -533,9 +533,11 @@ BackupRestore::object(Uint32 type, const void * ptr)
     if (!m_no_restore_disk)
     {
       NdbDictionary::Datafile old(*(NdbDictionary::Datafile*)ptr);
-      NdbDictionary::Tablespace * ts = m_tablespaces[old.getTablespaceId()];
+      NdbDictionary::ObjectId objid;
+      old.getTablespaceId(&objid);
+      NdbDictionary::Tablespace * ts = m_tablespaces[objid.getObjectId()];
       debug << "Connecting datafile " << old.getPath() 
-	    << " to tablespace: oldid: " << old.getTablespaceId() 
+	    << " to tablespace: oldid: " << objid.getObjectId()
 	    << " newid: " << ts->getObjectId() << endl;
       old.setTablespace(* ts);
       info << "Creating datafile \"" << old.getPath() << "\"..." << flush;
@@ -554,10 +556,11 @@ BackupRestore::object(Uint32 type, const void * ptr)
     if (!m_no_restore_disk)
     {
       NdbDictionary::Undofile old(*(NdbDictionary::Undofile*)ptr);
-      NdbDictionary::LogfileGroup * lg = 
-	m_logfilegroups[old.getLogfileGroupId()];
+      NdbDictionary::ObjectId objid;
+      old.getLogfileGroupId(&objid);
+      NdbDictionary::LogfileGroup * lg = m_logfilegroups[objid.getObjectId()];
       debug << "Connecting undofile " << old.getPath() 
-	    << " to logfile group: oldid: " << old.getLogfileGroupId() 
+	    << " to logfile group: oldid: " << objid.getObjectId()
 	    << " newid: " << lg->getObjectId() 
 	    << " " << (void*)lg << endl;
       old.setLogfileGroup(* lg);